adminandClaude Opus 4.8
be6909bce2
Edge /api/vehicles (RDW NL, live now) + /api/screen (OpenSanctions, live now), both defer to engine when ENGINE_URL set. Engine: /vehicles/{plate} + /screen endpoints; workers/resolve (blocking+fuzzy entity resolution, now used by pipeline); workers/queue (in-process job queue); sources/financials (Companies House accounts + XBRL path). py_compile + tsc + next build green.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
32 lines
861 B
Python
32 lines
861 B
Python
"""
|
|
Lightweight in-process job queue (docs/02). Adequate for single-worker dev; swap
|
|
the backend for Redis/Celery at scale without changing callers.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Any, Awaitable, Callable
|
|
|
|
Job = Callable[..., Awaitable[Any]]
|
|
|
|
|
|
class JobQueue:
|
|
def __init__(self) -> None:
|
|
self._q: "asyncio.Queue[tuple[Job, tuple[Any, ...]]]" = asyncio.Queue()
|
|
|
|
async def enqueue(self, fn: Job, *args: Any) -> None:
|
|
await self._q.put((fn, args))
|
|
|
|
async def worker(self) -> None:
|
|
while True:
|
|
fn, args = await self._q.get()
|
|
try:
|
|
await fn(*args)
|
|
except Exception as e: # keep the worker alive on a failed job
|
|
print("job failed:", repr(e))
|
|
finally:
|
|
self._q.task_done()
|
|
|
|
|
|
queue = JobQueue()
|