api/core/task_queue.py - master - systemsstrategyy/Treker

api/core/task_queue.py

42 строки2 KB

SystemsStrategy

Синхронизация с актуальной линией разработки (август 2026)

06 авг 2026, 12:52

06 авг 2026, 12:5219af6aa

"""Открытие Procrastinate App в API-процессе для defer_async из HTTP.

Worker открывает App сам (workers/procrastinate_worker.py). API — отдельный
процесс и тоже должен открыть connector, иначе defer_async из HTTP-обработчиков
(NotificationService.notify, auth.register → onboard_organization) бросает
AppNotOpen и задача молча теряется. Схему НЕ применяем — её ставит worker
(ensure_schema); здесь только открываем connection-pool для defer.

По образцу api/modules/notifications/api.py::setup_notifications: hook кладётся
в app.state.lifespan_hooks, который api/main.py итерирует через AsyncExitStack.
"""

from __future__ import annotations

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any

from fastapi import FastAPI


def setup_task_queue(app: FastAPI) -> None:
    """Регистрирует lifespan-hook открытия Procrastinate App. Идемпотентна."""
    if getattr(app.state, "_task_queue_registered", False):
        return
    app.state._task_queue_registered = True

    hooks: list[Any] | None = getattr(app.state, "lifespan_hooks", None)
    if hooks is None:
        hooks = []
        app.state.lifespan_hooks = hooks

    @asynccontextmanager
    async def _task_queue_lifespan(app_: FastAPI) -> AsyncIterator[None]:
        # Lazy-import: workers.queue на import создаёт PsycopgConnector (тянет
        # psycopg + читает settings.database.url) — не нужно на import api.main.
        from workers.queue import app as procrastinate_app

        async with procrastinate_app.open_async():
            yield

    hooks.append(_task_queue_lifespan)