Spaces:
Sleeping
Sleeping
File size: 1,846 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 | """Temporal worker entry point for report generation workflows."""
from __future__ import annotations
import asyncio
import logging
from app.config import settings
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
async def main() -> None:
try:
from temporalio.client import Client
from temporalio.worker import Worker
except ImportError as exc: # pragma: no cover
raise SystemExit(
"temporalio is required for the worker. Install: pip install 'report-genius-ai[temporal]'"
) from exc
from app.db.database import init_db
from app.services.generation_stale import generation_stale_sweeper_loop
await init_db()
if int(getattr(settings, "generation_stale_sweep_seconds", 120)) > 0:
asyncio.create_task(generation_stale_sweeper_loop())
from app.workflows.report_activities import (
assemble_report,
execute_report_generation,
fetch_sources,
generate_section_activity,
retrieve_context,
validate_output,
)
from app.workflows.report_workflow import ReportGenerationWorkflow
client = await Client.connect(
settings.temporal_host,
namespace=settings.temporal_namespace,
)
worker = Worker(
client,
task_queue=settings.temporal_task_queue,
workflows=[ReportGenerationWorkflow],
activities=[
fetch_sources,
retrieve_context,
generate_section_activity,
execute_report_generation,
validate_output,
assemble_report,
],
)
logger.info(
"Temporal worker started queue=%s host=%s",
settings.temporal_task_queue,
settings.temporal_host,
)
await worker.run()
if __name__ == "__main__":
asyncio.run(main())
|