From 19c2a378e7e66652afbfa3e749badd697e3917ac Mon Sep 17 00:00:00 2001 From: Florent Chehab Date: Mon, 30 Mar 2026 14:20:04 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8(summary)=20taskV2=20closer=20to=20tar?= =?UTF-8?q?get=20API=20gateway=20contract?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Updated taskV2 API contract to be closer to the target gateway contract. GET operations return the same things as the webhook payload. Also store the summary on S3 to be iso with transcript. --- .../dev-keycloak/values.meet.yaml.gotmpl | 2 +- src/summary/summary/api/route/tasks_v2.py | 84 +++++++++++++++---- src/summary/summary/core/celery_worker.py | 35 ++++---- src/summary/summary/core/config.py | 1 + src/summary/summary/core/file_service.py | 24 +++++- src/summary/summary/core/shared_models.py | 32 ++++++- src/summary/tests/api/test_api_tasks_v2.py | 67 ++++++++++++--- 7 files changed, 194 insertions(+), 51 deletions(-) diff --git a/src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl b/src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl index 8d8a51d2..6bb6fdf5 100644 --- a/src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl +++ b/src/helm/env.d/dev-keycloak/values.meet.yaml.gotmpl @@ -6,7 +6,7 @@ _summaryEnvVars: &summaryEnvVars AWS_S3_ACCESS_KEY_ID: meet AWS_S3_SECRET_ACCESS_KEY: password AWS_S3_SECURE_ACCESS: False - AUTHORIZED_TENANTS: '[{"id": "dictaphone", "api_key": "dictaphone_token", "webhook_url": "http://dictaphone-backend.dictaphone.svc.cluster.local/api/v1.0/files/ai-webhook/", "webhook_api_key": "token_summary"}]' + AUTHORIZED_TENANTS: '[{"id": "dictaphone", "api_key": "dictaphone_token", "webhook_url": "http://dictaphone-backend.dictaphone.svc.cluster.local/api/v1.0/ai-jobs/webhook/", "webhook_api_key": "token_summary"}]' WHISPERX_API_KEY: secretKeyRef: name: secret-dev diff --git a/src/summary/summary/api/route/tasks_v2.py b/src/summary/summary/api/route/tasks_v2.py index f53b5e9a..e8f8cc78 100644 --- a/src/summary/summary/api/route/tasks_v2.py +++ b/src/summary/summary/api/route/tasks_v2.py @@ -1,15 +1,24 @@ """API routes related to application tasks (V2 / tenant friendly).""" from celery.result import AsyncResult -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Request from summary.core.celery_worker import ( + celery, process_audio_transcribe_v2_task, summarize_v2_task, ) from summary.core.config import AuthorizedTenant from summary.core.models import SummarizeTaskV2Request, TranscribeTaskV2Request from summary.core.security import verify_tenant_api_key_v2 +from summary.core.shared_models import ( + SummarizeWebhookFailurePayload, + SummarizeWebhookPendingPayload, + SummarizeWebhookSuccessPayload, + TranscribeWebhookFailurePayload, + TranscribeWebhookPendingPayload, + TranscribeWebhookSuccessPayload, +) router_tasks_v2 = APIRouter() @@ -24,7 +33,7 @@ async def create_transcribe_task_v2( args=[{**request.model_dump(), "tenant_id": request_tenant.id}] ) - return {"job_id": task.id, "message": "Transcribe job created"} + return TranscribeWebhookPendingPayload(job_id=task.id).model_dump() @router_tasks_v2.post("/async-jobs/summarize") @@ -37,25 +46,64 @@ async def create_summarize_task_v2( args=[{**request.model_dump(), "tenant_id": request_tenant.id}] ) - return {"job_id": task.id, "message": "Summarize job created"} + return SummarizeWebhookPendingPayload(job_id=task.id).model_dump() @router_tasks_v2.get("/async-jobs/transcribe/{job_id}") -@router_tasks_v2.get("/async-jobs/summarize/{job_id}") -async def get_task_status( +async def get_transcribe_job_status( job_id: str, + request: Request, request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2), ): - """Check task status by ID.""" - task = AsyncResult(job_id) - try: - if ( - isinstance(task.args, (list, tuple)) - and len(task.args) > 0 - and isinstance(task.args[0], dict) - and task.args[0].get("tenant_id") == request_tenant.id - ): - return {"job_id": job_id, "status": task.status} - except (TypeError, KeyError): - pass - raise HTTPException(status_code=404, detail="Not found") + """Check transcription task status by ID.""" + # We have to look directly in Redis to check if the task exists + redis_client = celery.backend.client + key = f"celery-task-meta-{job_id}" + if not redis_client.exists(key): + raise HTTPException(status_code=404, detail="Not found") + + task = AsyncResult(job_id, app=celery) + task_tenant_id = task.args[0]["tenant_id"] + if task_tenant_id != request_tenant.id: + raise HTTPException(status_code=403, detail="Forbidden") + + if task.status == "SUCCESS": + result = task.result + return TranscribeWebhookSuccessPayload.model_validate(result).model_dump() + + if task.status == "FAILURE": + return TranscribeWebhookFailurePayload( + job_id=job_id, error_code="unknown_error" + ).model_dump() + + return TranscribeWebhookPendingPayload(job_id=job_id).model_dump() + + +@router_tasks_v2.get("/async-jobs/summarize/{job_id}") +async def get_summarize_job_status( + job_id: str, + request: Request, + request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2), +): + """Check summarize task status by ID.""" + # We have to look directly in Redis to check if the task exists + redis_client = celery.backend.client + key = f"celery-task-meta-{job_id}" + if not redis_client.exists(key): + raise HTTPException(status_code=404, detail="Not found") + + task = AsyncResult(job_id, app=celery) + task_tenant_id = task.args[0]["tenant_id"] + if task_tenant_id != request_tenant.id: + raise HTTPException(status_code=403, detail="Forbidden") + + if task.status == "SUCCESS": + result = task.result + return SummarizeWebhookSuccessPayload.model_validate(result).model_dump() + + if task.status == "FAILURE": + return SummarizeWebhookFailurePayload( + job_id=job_id, error_code="unknown_error" + ).model_dump() + + return SummarizeWebhookPendingPayload(job_id=job_id).model_dump() diff --git a/src/summary/summary/core/celery_worker.py b/src/summary/summary/core/celery_worker.py index 7a88aae3..7c0b8e4d 100644 --- a/src/summary/summary/core/celery_worker.py +++ b/src/summary/summary/core/celery_worker.py @@ -58,6 +58,9 @@ celery = Celery( broker=settings.celery_broker_url, backend=settings.celery_result_backend, broker_connection_retry_on_startup=True, + # To store the tasks args too in results and make the + # V2 API work + result_extended=True, ) celery.config_from_object("summary.core.celery_config") @@ -465,15 +468,14 @@ def process_audio_transcribe_v2_task( job_id=job_id, ) - call_webhook_v2_task.apply_async( - args=[ - TranscribeWebhookSuccessPayload( - job_id=job_id, - transcription_data_url=file_service.get_transcript_signed_url(job_id), - ).model_dump(), - payload.tenant_id, - ] + success_payload = TranscribeWebhookSuccessPayload( + job_id=job_id, + transcription_data_url=file_service.get_transcript_signed_url(job_id), ) + call_webhook_v2_task.apply_async( + args=[success_payload.model_dump(), payload.tenant_id] + ) + return success_payload.model_dump() @signals.task_failure.connect(sender=process_audio_transcribe_v2_task) @@ -531,14 +533,17 @@ def summarize_v2_task( transcript=payload.content, session_id=self.request.id, ) - call_webhook_v2_task.apply_async( - args=[ - SummarizeWebhookSuccessPayload( - job_id=self.request.id, summary=summary - ).model_dump(), - payload.tenant_id, - ] + job_id = self.request.id + file_service.store_summary(summary=summary, job_id=job_id) + + success_payload = SummarizeWebhookSuccessPayload( + job_id=job_id, + summary_data_url=file_service.get_summary_signed_url(job_id), ) + call_webhook_v2_task.apply_async( + args=[success_payload.model_dump(), payload.tenant_id] + ) + return success_payload.model_dump() @signals.task_failure.connect(sender=summarize_v2_task) diff --git a/src/summary/summary/core/config.py b/src/summary/summary/core/config.py index 5b0096af..74753ab3 100644 --- a/src/summary/summary/core/config.py +++ b/src/summary/summary/core/config.py @@ -78,6 +78,7 @@ class Settings(BaseSettings): aws_s3_secret_access_key: SecretStr aws_s3_secure_access: bool = True aws_transcript_path: str = "transcripts" + aws_summary_path: str = "summaries" # AI-related settings whisperx_api_key: SecretStr diff --git a/src/summary/summary/core/file_service.py b/src/summary/summary/core/file_service.py index 2f945c71..45636fe0 100644 --- a/src/summary/summary/core/file_service.py +++ b/src/summary/summary/core/file_service.py @@ -293,5 +293,27 @@ class FileService: transcript_path = f"{settings.aws_transcript_path}/{job_id}.json" logger.debug("Transcript path: %s", transcript_path) return self._minio_client.presigned_get_object( - self._bucket_name, transcript_path, expires=timedelta(hours=1) + self._bucket_name, transcript_path, expires=timedelta(hours=24) + ) + + def store_summary(self, *, summary: str, job_id: str) -> None: + """Store summary in MinIO.""" + logger.info("Storing summary for job id %s", job_id) + summary_path = f"{settings.aws_summary_path}/{job_id}.txt" + logger.debug("Summary path: %s", summary_path) + data = summary.encode() + self._minio_client.put_object( + self._bucket_name, + summary_path, + io.BytesIO(data), + length=len(data), + ) + logger.info("Summary stored successfully for job id %s", job_id) + + def get_summary_signed_url(self, job_id: str) -> str: + """Get signed URL for summary file.""" + summary_path = f"{settings.aws_summary_path}/{job_id}.txt" + logger.debug("Summary path: %s", summary_path) + return self._minio_client.presigned_get_object( + self._bucket_name, summary_path, expires=timedelta(hours=24) ) diff --git a/src/summary/summary/core/shared_models.py b/src/summary/summary/core/shared_models.py index f213da2c..340a9959 100644 --- a/src/summary/summary/core/shared_models.py +++ b/src/summary/summary/core/shared_models.py @@ -65,6 +65,13 @@ class TranscribeWebhookSuccessPayload(BaseWebhook): ) +class TranscribeWebhookPendingPayload(BaseWebhook): + """Payload for a pending transcription webhook-like response.""" + + type: Literal["transcript"] = Field(default="transcript") + status: Literal["pending"] = Field(default="pending") + + class TranscribeWebhookFailurePayload(BaseWebhook): """Payload for a failed transcription webhook.""" @@ -76,7 +83,11 @@ class TranscribeWebhookFailurePayload(BaseWebhook): TranscribeWebhookPayloads = Annotated[ - Union[TranscribeWebhookSuccessPayload, TranscribeWebhookFailurePayload], + Union[ + TranscribeWebhookSuccessPayload, + TranscribeWebhookPendingPayload, + TranscribeWebhookFailurePayload, + ], Field(discriminator="status"), ] @@ -86,7 +97,16 @@ class SummarizeWebhookSuccessPayload(BaseWebhook): type: Literal["summary"] = Field(default="summary") status: Literal["success"] = Field(default="success") - summary: str = Field(title="Summary", description="The summary of the text.") + summary_data_url: str = Field( + title="Summary", description="URL to the raw summary data." + ) + + +class SummarizeWebhookPendingPayload(BaseWebhook): + """Payload for a pending summarization webhook-like response.""" + + type: Literal["summary"] = Field(default="summary") + status: Literal["pending"] = Field(default="pending") class SummarizeWebhookFailurePayload(BaseWebhook): @@ -100,7 +120,11 @@ class SummarizeWebhookFailurePayload(BaseWebhook): SummarizeWebhookPayloads = Annotated[ - Union[SummarizeWebhookSuccessPayload, SummarizeWebhookFailurePayload], + Union[ + SummarizeWebhookSuccessPayload, + SummarizeWebhookPendingPayload, + SummarizeWebhookFailurePayload, + ], Field(discriminator="status"), ] @@ -114,8 +138,10 @@ webhook_payload_adapter = TypeAdapter(WebhookPayloads) __all__ = [ "TranscribeWebhookSuccessPayload", + "TranscribeWebhookPendingPayload", "TranscribeWebhookFailurePayload", "SummarizeWebhookSuccessPayload", + "SummarizeWebhookPendingPayload", "SummarizeWebhookFailurePayload", "TranscribeWebhookPayloads", "SummarizeWebhookPayloads", diff --git a/src/summary/tests/api/test_api_tasks_v2.py b/src/summary/tests/api/test_api_tasks_v2.py index ffde24ad..0fac599e 100644 --- a/src/summary/tests/api/test_api_tasks_v2.py +++ b/src/summary/tests/api/test_api_tasks_v2.py @@ -1,6 +1,6 @@ """Integration tests for the V2 task API endpoints.""" -from unittest.mock import MagicMock, patch +from unittest.mock import ANY, MagicMock, patch class TestTasksV2: @@ -26,7 +26,8 @@ class TestTasksV2: assert response.status_code == 200 assert response.json() == { "job_id": "transcribe-task-id-abc", - "message": "Transcribe job created", + "type": "transcript", + "status": "pending", } args = mock_apply_async.call_args.kwargs["args"] @@ -58,7 +59,8 @@ class TestTasksV2: assert response.status_code == 200 assert response.json() == { "job_id": "summarize-task-id-abc", - "message": "Summarize job created", + "type": "summary", + "status": "pending", } args = mock_apply_async.call_args.kwargs["args"] @@ -70,11 +72,13 @@ class TestTasksV2: } ] + @patch("summary.api.route.tasks_v2.celery") @patch("summary.api.route.tasks_v2.AsyncResult") def test_get_transcribe_task_status_returns_status_for_same_tenant( - self, mock_async_result, client + self, mock_async_result, mock_celery, client ): """GET /async-jobs/transcribe/{id} returns status when tenant matches.""" + mock_celery.backend.client.exists.return_value = True mock_async_result.return_value = MagicMock( status="PENDING", args=[{"tenant_id": "test-tenant"}], @@ -86,18 +90,28 @@ class TestTasksV2: ) assert response.status_code == 200 - assert response.json() == {"job_id": "task-id-abc", "status": "PENDING"} + assert response.json() == { + "job_id": "task-id-abc", + "type": "transcript", + "status": "pending", + } - mock_async_result.assert_called_once_with("task-id-abc") + mock_async_result.assert_called_once_with("task-id-abc", app=ANY) + @patch("summary.api.route.tasks_v2.celery") @patch("summary.api.route.tasks_v2.AsyncResult") def test_get_summarize_task_status_returns_status_for_same_tenant( - self, mock_async_result, client + self, mock_async_result, mock_celery, client ): """GET /async-jobs/summarize/{id} returns status when tenant matches.""" + mock_celery.backend.client.exists.return_value = True mock_async_result.return_value = MagicMock( status="SUCCESS", args=[{"tenant_id": "test-tenant"}], + result={ + "job_id": "task-id-abc", + "summary_data_url": "https://example.com/summary.json", + }, ) response = client.get( @@ -106,16 +120,22 @@ class TestTasksV2: ) assert response.status_code == 200 - assert response.json() == {"job_id": "task-id-abc", "status": "SUCCESS"} + assert response.json() == { + "job_id": "task-id-abc", + "type": "summary", + "status": "success", + "summary_data_url": "https://example.com/summary.json", + } - mock_async_result.assert_called_once_with("task-id-abc") + mock_async_result.assert_called_once_with("task-id-abc", app=ANY) + @patch("summary.api.route.tasks_v2.celery") @patch("summary.api.route.tasks_v2.AsyncResult") - def test_get_task_status_returns_404_when_args_are_missing( - self, mock_async_result, client + def test_get_task_status_returns_404_when_job_does_not_exist( + self, mock_async_result, mock_celery, client ): - """GET /async-jobs/.../{id} returns 404 when task args are invalid.""" - mock_async_result.return_value = MagicMock(status="SUCCESS", args=None) + """GET /async-jobs/.../{id} returns 404 when task key is not in Redis.""" + mock_celery.backend.client.exists.return_value = False response = client.get( "/api/v2/async-jobs/transcribe/task-id-abc", @@ -124,3 +144,24 @@ class TestTasksV2: assert response.status_code == 404 assert response.json() == {"detail": "Not found"} + mock_async_result.assert_not_called() + + @patch("summary.api.route.tasks_v2.celery") + @patch("summary.api.route.tasks_v2.AsyncResult") + def test_get_task_status_returns_403_for_other_tenant( + self, mock_async_result, mock_celery, client + ): + """GET /async-jobs/.../{id} returns 403 when task belongs to another tenant.""" + mock_celery.backend.client.exists.return_value = True + mock_async_result.return_value = MagicMock( + status="PENDING", + args=[{"tenant_id": "another-tenant"}], + ) + + response = client.get( + "/api/v2/async-jobs/summarize/task-id-abc", + headers={"Authorization": "Bearer test-api-token"}, + ) + + assert response.status_code == 403 + assert response.json() == {"detail": "Forbidden"}