diff --git a/api/src/app.py b/api/src/app.py index 9d0922e..b3e2592 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -1,20 +1,17 @@ """Точка входа FastAPI сервиса tabular-parser. -Что здесь есть: настройки, подключение S3 (MinIO), NATS (nkeys JWT; пустые seed/jwt — без аутентификации), -middleware (gzip, proxy headers), healthcheck, обработчики исключений. Сервис без БД и фронта (DECISION-4). -Воркер обработки `recognition.request` и поле `nats` в healthcheck добавляются позже (src/core/nats.py). +Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без +аутентификации, фоновое подключение — старт не ждёт NATS), middleware (gzip, proxy headers), healthcheck, обработчики +исключений. Сервис без БД и фронта (DECISION-4). Воркер обработки `recognition.tabular.request` и поле `nats` +в healthcheck добавляются в задаче 4.3. """ -import base64 from contextlib import asynccontextmanager from datetime import UTC, datetime -from typing import Any -import nkeys from fastapi import FastAPI from loguru import logger from miniopy_async import Minio -from nats.aio.client import Client as NATS from starlette.middleware.gzip import GZipMiddleware from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware @@ -23,6 +20,7 @@ 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 settings = get_settings() @@ -32,49 +30,6 @@ UP_TIME = datetime.now(UTC) setup_logger(level="DEBUG" if settings.DEBUG else "INFO") -async def _connect_nats() -> NATS | None: - """Подключение к NATS (core, без JetStream); при ошибке — ``None``, приложение всё равно стартует.""" - nats_settings = settings.OPENBAO.NATS - - async def disconnected_cb() -> None: - logger.warning("NATS: соединение потеряно") - - async def reconnected_cb() -> None: - logger.info("NATS: соединение восстановлено") - - async def closed_cb() -> None: - logger.warning("NATS: соединение закрыто") - - # -1 — бесконечное автопереподключение (D-6): иначе клиент закрывается навсегда после короткого сбоя NATS - options: dict[str, Any] = { - "max_reconnect_attempts": -1, - "disconnected_cb": disconnected_cb, - "reconnected_cb": reconnected_cb, - "closed_cb": closed_cb, - } - if nats_settings.NATS_SEED and nats_settings.NATS_JWT: - kp = nkeys.from_seed(nats_settings.NATS_SEED.encode("utf-8")) - - def sig_cb(nonce: bytes | str) -> bytes: - if isinstance(nonce, str): - nonce = nonce.encode("utf-8") - return base64.b64encode(kp.sign(nonce)) - - options |= {"user_jwt_cb": lambda: nats_settings.NATS_JWT.encode("utf-8"), "signature_cb": sig_cb} - nc = NATS() - try: - await nc.connect(nats_settings.NATS_URL, **options) - except Exception as exc: - logger.exception( - "Не удалось подключиться к NATS. error_category=nats_error, error_type={error_type}, error={error}", - error_type=type(exc).__name__, - error=repr(exc), - ) - return None - logger.info("Подключение к NATS произошло") - return nc - - async def _connect_s3() -> S3Storage | None: """Клиент S3 с проверкой бакета; при недоступности — ``None`` (приложение всё равно стартует).""" minio = settings.OPENBAO.MINIO @@ -102,8 +57,12 @@ async def _connect_s3() -> S3Storage | None: @asynccontextmanager async def lifespan(app: FastAPI): app.state.storage = await _connect_s3() - app.state.nats = await _connect_nats() if settings.NATS_ENABLED else None - if not settings.NATS_ENABLED: + app.state.nats = None + if settings.NATS_ENABLED: + # start() не ждёт NATS: при недоступном сервере подключение продолжается в фоне (design D-6) + app.state.nats = NatsConnection(settings.OPENBAO.NATS, publish_retries=settings.PUBLISH_RETRIES) + await app.state.nats.start() + else: logger.info("NATS выключен (NATS_ENABLED=false)") try: @@ -112,7 +71,7 @@ async def lifespan(app: FastAPI): finally: logger.info("Приложение закрывается") if app.state.nats: - await app.state.nats.close() + await app.state.nats.stop() if app.state.storage: await app.state.storage.client.close_session()