diff --git a/solution/services/backend/api/v1/generate/__init__.py b/solution/services/backend/api/v1/generate/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/solution/services/backend/api/v1/generate/schemas.py b/solution/services/backend/api/v1/generate/schemas.py new file mode 100644 index 0000000..66326e2 --- /dev/null +++ b/solution/services/backend/api/v1/generate/schemas.py @@ -0,0 +1,14 @@ +from typing import Any + +from ninja import Schema + + +class GenerateAdTextIn(Schema): + advertiser_name: str + ad_title: str + + +class Promise(Schema): + task_id: str + status: str + result: Any diff --git a/solution/services/backend/api/v1/generate/views.py b/solution/services/backend/api/v1/generate/views.py new file mode 100644 index 0000000..1b25373 --- /dev/null +++ b/solution/services/backend/api/v1/generate/views.py @@ -0,0 +1,56 @@ +from http import HTTPStatus as status +from uuid import UUID + +import celery.states +from celery.result import AsyncResult +from django.http import Http404, HttpRequest +from ninja import Router + +from api.v1 import schemas as global_schemas +from api.v1.generate import schemas +from apps.campaign.tasks import generate_ad_text_task + +router = Router(tags=["generate"]) + + +@router.post( + "/ad_text", + response={ + status.OK: schemas.Promise, + status.BAD_REQUEST: global_schemas.BadRequestError, + }, +) +def generate_ad_text( + request: HttpRequest, prompt: schemas.GenerateAdTextIn +) -> tuple[status, schemas.Promise]: + task = generate_ad_text_task.delay(prompt.advertiser_name, prompt.ad_title) + task_result = AsyncResult(task.id) + + return status.OK, schemas.Promise( + task_id=task.id, + status=task_result.status, + result=task_result.result, + ) + + +@router.get( + "/ad_text/{task_id}/result", + response={ + status.OK: schemas.Promise, + status.BAD_REQUEST: global_schemas.BadRequestError, + status.NOT_FOUND: global_schemas.NotFoundError, + }, +) +def get_generate_ad_text_result( + request: HttpRequest, task_id: UUID +) -> tuple[status, schemas.Promise]: + task_result = AsyncResult(str(task_id)) + + if task_result.status == celery.states.PENDING: + raise Http404 + + return status.OK, schemas.Promise( + task_id=task_result.task_id, + status=task_result.status, + result=task_result.result, + )