Chris CoutinhoandClaude Opus 4.8
b10ce15032
🟡 Document the _doc_queueing_lock ":" delimiter invariant (user_id and the controlled doc_type enum are colon-free, so the key is collision-safe; a future doc_type with ":" must not be added). 🟡 API pod no longer opens the procrastinate connector twice on startup: add ProcrastinateTaskProducer.ensure_schema() (applies the schema on the already-open pool) and have both lifespan branches build the producer then ensure_schema — one open/close cycle, matching the worker. build_producer now returns the concrete producer type. 🟢 Document in ports.py that a long-lived-connection producer may optionally provide drain() (lifespan probes via getattr). 🟢 Add a unit test that a non-credential pipeline error propagates (for procrastinate's RetryStrategy) and still closes the client via finally. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
43 lines
1.5 KiB
Python
43 lines
1.5 KiB
Python
"""Composition root for the ingest producer (Deck #183).
|
|
|
|
The transport is selected from ``INGEST_QUEUE``:
|
|
|
|
- ``postgres`` → :class:`ProcrastinateTaskProducer`, which defers jobs into the
|
|
per-tenant Postgres for the out-of-process ``worker`` role to drain.
|
|
- ``memory`` (SQLite/dev default) → the in-process anyio stream, built inline by
|
|
the server lifespan (it owns both the send and receive ends), so it is not
|
|
produced here.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from ...config import Settings
|
|
|
|
if TYPE_CHECKING:
|
|
from .procrastinate import ProcrastinateTaskProducer
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def build_producer(settings: Settings) -> ProcrastinateTaskProducer:
|
|
"""Build the Postgres (procrastinate) ingest producer.
|
|
|
|
Returns the concrete :class:`ProcrastinateTaskProducer` (not just the
|
|
``TaskProducer`` protocol) so the lifespan can call ``ensure_schema()`` on
|
|
the open connector. Precondition: ``settings.ingest_queue == "postgres"``
|
|
(the memory transport is constructed inline by the lifespan because it needs
|
|
the paired receive stream for the in-process processor pool).
|
|
"""
|
|
if settings.ingest_queue != "postgres":
|
|
raise ValueError(
|
|
"build_producer is only for INGEST_QUEUE=postgres; the memory "
|
|
f"transport is built inline by the lifespan (got {settings.ingest_queue!r})"
|
|
)
|
|
|
|
from .procrastinate import ProcrastinateTaskProducer # noqa: PLC0415
|
|
|
|
return await ProcrastinateTaskProducer.connect()
|