"""Точка входа FastAPI сервиса tabular-parser. Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без аутентификации, фоновое подключение — старт не ждёт NATS), воркер `src/core/worker.py` (подписка на `recognition.tabular.request` после подключения к 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 from src.core.worker import STOP_TIMEOUT, RecognitionWorker from src.parser import get_parser 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, concurrency=settings.WORKER_CONCURRENCY, done_cache_size=settings.DONE_CACHE_SIZE, ) return nats, worker @asynccontextmanager async def lifespan(app: FastAPI): 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)} app.include_router(router) register_exception_handlers(app) app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=1) app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"])