Vishesh 'ironeagle' Bangotra
0cb35d10a6
- build.py: _TEMPLATE_COMPONENTS rewritten around the vet sample (53 kinds —
per-resource slices, jobs, worked appointment_reminder family, tests);
seed tokens dropped from the manifest ('sample: vet' instead)
- registry.py: get_app_component(kind) no longer renames item/example seeds;
prompts and tool docs updated to 'copy a slice and adapt'
- templates/README.md, 01_build_recipe, 40_ui_admin, wiki/03 updated off the
old seed scheme; repo inventory/config now exclude vet (docs move under
standards; react/vet itself is deleted in a later phase)
- regenerated MCP bundle, vendored templates, and site output
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""# Summary
|
|
|
|
Background worker — polls for pending jobs and runs the matching pipeline
|
|
family. The poll loop is the pipeline axis of the sample: resources are created
|
|
by the API axis, workers consume them on a fixed cadence.
|
|
|
|
Notes:
|
|
- This process is separate from the API process. Any in-memory event bus is per
|
|
process — if the pipeline must surface live progress to the API, swap the bus
|
|
adaptor for a broker-backed one (see standards 51_worker_and_progress).
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
from typing import Any, Awaitable, Callable
|
|
|
|
from config import settings
|
|
from pipelines.appointment_reminder.impl import get_appointment_reminder_pipeline
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# job.kind -> async (job, resources) -> None. Add families here as they land.
|
|
HANDLERS: dict[str, Callable[..., Awaitable[None]]] = {}
|
|
|
|
|
|
def _register_default_handlers() -> None:
|
|
"""Populate the dispatch table with the shipped families."""
|
|
|
|
async def run_appointment_reminder(job, resources: dict[str, Any]) -> None:
|
|
payload = job.payload or {}
|
|
appointment_id = payload.get("appointment_id")
|
|
if appointment_id is None:
|
|
raise ValueError("appointment_reminder job requires payload.appointment_id")
|
|
pipeline = get_appointment_reminder_pipeline(resources["appointment_svc"])
|
|
await pipeline.run(int(appointment_id))
|
|
|
|
HANDLERS["appointment_reminder"] = run_appointment_reminder
|
|
|
|
|
|
_register_default_handlers()
|
|
|
|
|
|
async def poll_jobs(resources: dict) -> None:
|
|
"""Poll for pending jobs and dispatch each to its pipeline family.
|
|
|
|
Args:
|
|
resources (dict): Wired services from :func:`init.init`.
|
|
"""
|
|
job_svc = resources["job_svc"]
|
|
logger.info("WORKER STARTED — poll=%ss max_retries=%d",
|
|
settings.worker_poll_seconds, settings.worker_max_retries)
|
|
|
|
while True:
|
|
pending = await job_svc.search({"status": "pending"})
|
|
pending = pending[:10]
|
|
logger.info("Found %d pending job(s)", len(pending))
|
|
|
|
for job in pending:
|
|
job_id = str(job.id)
|
|
if (job.retry_count or 0) >= settings.worker_max_retries:
|
|
logger.warning("SKIPPING job %s — retries exhausted", job_id)
|
|
continue
|
|
|
|
handler = HANDLERS.get(job.kind)
|
|
await job_svc.patch(job_id, {"status": "processing"})
|
|
try:
|
|
if handler is None:
|
|
raise ValueError(f"no handler for job kind '{job.kind}'")
|
|
await handler(job, resources)
|
|
await job_svc.patch(job_id, {"status": "done", "error_message": ""})
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.exception("job %s FAILED: %s", job_id, exc)
|
|
await job_svc.patch(job_id, {
|
|
"status": "failed",
|
|
"error_message": str(exc),
|
|
"retry_count": (job.retry_count or 0) + 1,
|
|
})
|
|
|
|
await asyncio.sleep(settings.worker_poll_seconds)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
from init import init, shutdown
|
|
|
|
async def _run():
|
|
resources = await init()
|
|
try:
|
|
await poll_jobs(resources)
|
|
finally:
|
|
await shutdown()
|
|
|
|
asyncio.run(_run())
|