✨(summary) taskV2 closer to target API gateway contract

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.

Author: FloChehab

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

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}")
+async def get_transcribe_job_status(
+    job_id: str,
+    request: Request,
+    request_tenant: AuthorizedTenant = Depends(verify_tenant_api_key_v2),
+):
+    """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_task_status(
+async def get_summarize_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 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()

src/summary/summary/core/celery_worker.py

@@ -58,6 +58,9 @@
     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,
     )
 
+    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=[
-            TranscribeWebhookSuccessPayload(
-                job_id=job_id,
-                transcription_data_url=file_service.get_transcript_signed_url(job_id),
-            ).model_dump(),
-            payload.tenant_id,
-        ]
+        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,
     )
+    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=[
-            SummarizeWebhookSuccessPayload(
-                job_id=self.request.id, summary=summary
-            ).model_dump(),
-            payload.tenant_id,
-        ]
+        args=[success_payload.model_dump(), payload.tenant_id]
     )
+    return success_payload.model_dump()
 
 
 @signals.task_failure.connect(sender=summarize_v2_task)

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

src/summary/summary/core/file_service.py

@@ -293,5 +293,27 @@ def get_transcript_signed_url(self, job_id: str) -> str:
         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)
         )

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 @@ class SummarizeWebhookFailurePayload(BaseWebhook):
 
 __all__ = [
     "TranscribeWebhookSuccessPayload",
+    "TranscribeWebhookPendingPayload",
     "TranscribeWebhookFailurePayload",
     "SummarizeWebhookSuccessPayload",
+    "SummarizeWebhookPendingPayload",
     "SummarizeWebhookFailurePayload",
     "TranscribeWebhookPayloads",
     "SummarizeWebhookPayloads",