arm-tabular-parser/api/src/app.py

"""Точка входа FastAPI сервиса tabular-parser.

Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без
аутентификации, фоновое подключение — старт не ждёт NATS), воркер `src/core/worker.py` (подписка на
subject запроса `tabular` из настроек после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck`
(`nats = ok | down | off`, R-17), обработчики исключений. Сервис без БД и фронта (DECISION-4).

При `NATS_ENABLED=false` ни NATS, ни воркер не запускаются. Остановка: воркер перестаёт принимать запросы и
дорабатывает очередь не дольше `STOP_TIMEOUT` (NATS ещё нужен для публикации результатов), затем закрывается NATS
(drain не дольше `DRAIN_TIMEOUT`), затем сессия S3 — в сумме в пределах 30 с, которые платформа даёт на завершение.
"""

from contextlib import asynccontextmanager
from datetime import UTC, datetime

from fastapi import FastAPI, Request
from loguru import logger
from miniopy_async import Minio
from starlette.middleware.gzip import GZipMiddleware
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware

from src.api.router import router
from src.core.clients import S3Storage
from src.core.config import get_settings
from src.core.exception_handlers import register_exception_handlers
from src.core.logger import setup_logger
from src.core.nats import NatsConnection, used_subjects
from src.core.worker import STOP_TIMEOUT, RecognitionWorker
from src.parser import get_parser, require_statements_version

settings = get_settings()

VERSION = settings.VERSION
UP_TIME = datetime.now(UTC)

setup_logger(level="DEBUG" if settings.DEBUG else "INFO")


async def _connect_s3() -> S3Storage:
    """Клиент S3 с проверкой бакета.

    Проверка — только диагностика: при недоступном MinIO или отсутствующем бакете ошибка логируется, а клиент
    остаётся (соединения открываются на каждый запрос) — когда MinIO поднимется, воркер прочитает файл без рестарта.
    """
    minio = settings.OPENBAO.MINIO
    s3_client = Minio(
        minio.MINIO_URL,
        access_key=minio.MINIO_ADMIN_LOGIN,
        secret_key=minio.MINIO_ADMIN_PASSWORD,
        secure=minio.MINIO_SECURE,
        cert_check=False,
    )
    try:
        bucket_exists = await s3_client.bucket_exists(minio.MINIO_BUCKET)
    except Exception as exc:
        logger.exception(
            "Не удалось проверить наличие бакета. error_category=s3_error, error_type={error_type}, error={error}",
            error_type=type(exc).__name__,
            error=repr(exc),
        )
    else:
        if bucket_exists:
            logger.info("Бакет проверен. bucket={bucket}", bucket=minio.MINIO_BUCKET)
        else:
            logger.warning(
                "Бакет не найден: чтение исходников и запись результатов будут падать. error_category=s3_error, "
                "bucket={bucket}",
                bucket=minio.MINIO_BUCKET,
            )
    return S3Storage(client=s3_client, bucket_name=minio.MINIO_BUCKET)


def _build_worker(storage: S3Storage) -> tuple[NatsConnection, RecognitionWorker]:
    """Соединение NATS и воркер; подписка на запросы — в `on_connected` (после первого подключения)."""

    async def subscribe(conn: NatsConnection) -> None:
        await conn.subscribe_requests(worker.on_request)

    nats = NatsConnection(settings.OPENBAO.NATS, on_connected=subscribe, publish_retries=settings.PUBLISH_RETRIES)
    worker = RecognitionWorker(
        storage=storage,
        publisher=nats,
        parser=get_parser(settings),
        service_name=settings.SERVICE_NAME,
        service_version=VERSION,
        statements_version=require_statements_version(),
        concurrency=settings.WORKER_CONCURRENCY,
        done_cache_size=settings.DONE_CACHE_SIZE,
        result_subject=settings.OPENBAO.NATS.NATS_RESULT_SUBJECT,
    )
    return nats, worker


@asynccontextmanager
async def lifespan(app: FastAPI):
    logger.info("Subject'ы NATS. {subjects}", subjects=used_subjects(settings.OPENBAO.NATS))
    storage = await _connect_s3()
    nats: NatsConnection | None = None
    worker: RecognitionWorker | None = None
    if settings.NATS_ENABLED:
        nats, worker = _build_worker(storage)
        worker.start()
        # start() не ждёт NATS: при недоступном сервере подключение продолжается в фоне (design D-6)
        await nats.start()
    else:
        logger.info("NATS выключен (NATS_ENABLED=false): воркер не запускается")
    app.state.storage = storage
    app.state.nats = nats
    app.state.worker = worker

    try:
        logger.info("Приложение запущено")
        yield
    finally:
        logger.info("Приложение закрывается")
        if worker is not None:
            await worker.stop(STOP_TIMEOUT)  # результаты ещё публикуются: NATS закрывается после воркера
        if nats is not None:
            await nats.stop()
        await storage.client.close_session()


app = FastAPI(title="tabular-parser", version=VERSION, root_path=settings.ROOT_PATH, lifespan=lifespan)


def _nats_state(application: FastAPI) -> str:
    """`off` — NATS выключен флагом (R-17); `ok` — соединение есть; `down` — включён, но связи нет."""
    if not settings.NATS_ENABLED:
        return "off"
    nats = getattr(application.state, "nats", None)
    return "ok" if nats is not None and nats.is_connected else "down"


@app.get("/healthcheck", status_code=200)
async def healthcheck(request: Request):
    return {
        "uptime": UP_TIME,
        "version": VERSION,
        "nats": _nats_state(request.app),
        "subjects": used_subjects(settings.OPENBAO.NATS),
    }


app.include_router(router)

register_exception_handlers(app)

app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=1)
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"])