Spaces:
Runtime error
Runtime error
File size: 2,234 Bytes
732b14f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | """Temporal client helpers for starting report generation workflows."""
from __future__ import annotations
import logging
from typing import Any
from app.config import settings
from app.workflows.models import ReportWorkflowRequest
logger = logging.getLogger(__name__)
_client: Any = None
async def get_temporal_client() -> Any:
global _client
if _client is not None:
return _client
try:
from temporalio.client import Client
except ImportError as exc: # pragma: no cover
raise RuntimeError("temporalio is not installed; pip install 'report-genius-ai[temporal]'") from exc
_client = await Client.connect(
settings.temporal_host,
namespace=settings.temporal_namespace,
)
return _client
async def start_report_generation_workflow(request: ReportWorkflowRequest) -> str:
"""Start ``ReportGenerationWorkflow``; returns workflow id."""
from app.workflows.report_workflow import ReportGenerationWorkflow
client = await get_temporal_client()
import uuid
wf_id = f"report-gen-{request.report_id}-{uuid.uuid4().hex[:12]}"
handle = await client.start_workflow(
ReportGenerationWorkflow.run,
request,
id=wf_id,
task_queue=settings.temporal_task_queue,
)
return str(handle.id)
async def try_start_report_generation_workflow(
request: ReportWorkflowRequest,
) -> str | None:
"""Return Temporal workflow id when start succeeded, else ``None`` (caller uses in-process task)."""
if not settings.enable_temporal_workflow:
return None
try:
from app.workflows.report_activities import temporal_activities_enabled
if not temporal_activities_enabled():
logger.warning(
"Temporal enabled but temporalio unavailable — falling back to in-process "
"generation report=%s",
request.report_id,
)
return None
return await start_report_generation_workflow(request)
except Exception: # noqa: BLE001
logger.exception(
"Temporal workflow start failed report=%s — falling back to in-process task",
request.report_id,
)
return None
|