refactor: lifespan подключается к NATS через core/nats.py (фоново, старт не ждёт NATS)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-10-04 02:47:47 +04:00
parent ff08905580
commit b545143f9e

View File

@ -1,20 +1,17 @@
"""Точка входа FastAPI сервиса tabular-parser. """Точка входа FastAPI сервиса tabular-parser.
Что здесь есть: настройки, подключение S3 (MinIO), NATS (nkeys JWT; пустые seed/jwt — без аутентификации), Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без
middleware (gzip, proxy headers), healthcheck, обработчики исключений. Сервис без БД и фронта (DECISION-4). аутентификации, фоновое подключение — старт не ждёт NATS), middleware (gzip, proxy headers), healthcheck, обработчики
Воркер обработки `recognition.request` и поле `nats` в healthcheck добавляются позже (src/core/nats.py). исключений. Сервис без БД и фронта (DECISION-4). Воркер обработки `recognition.tabular.request` и поле `nats`
в healthcheck добавляются в задаче 4.3.
""" """
import base64
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from datetime import UTC, datetime from datetime import UTC, datetime
from typing import Any
import nkeys
from fastapi import FastAPI from fastapi import FastAPI
from loguru import logger from loguru import logger
from miniopy_async import Minio from miniopy_async import Minio
from nats.aio.client import Client as NATS
from starlette.middleware.gzip import GZipMiddleware from starlette.middleware.gzip import GZipMiddleware
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware 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.config import get_settings
from src.core.exception_handlers import register_exception_handlers from src.core.exception_handlers import register_exception_handlers
from src.core.logger import setup_logger from src.core.logger import setup_logger
from src.core.nats import NatsConnection
settings = get_settings() settings = get_settings()
@ -32,49 +30,6 @@ UP_TIME = datetime.now(UTC)
setup_logger(level="DEBUG" if settings.DEBUG else "INFO") 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: async def _connect_s3() -> S3Storage | None:
"""Клиент S3 с проверкой бакета; при недоступности — ``None`` (приложение всё равно стартует).""" """Клиент S3 с проверкой бакета; при недоступности — ``None`` (приложение всё равно стартует)."""
minio = settings.OPENBAO.MINIO minio = settings.OPENBAO.MINIO
@ -102,8 +57,12 @@ async def _connect_s3() -> S3Storage | None:
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
app.state.storage = await _connect_s3() app.state.storage = await _connect_s3()
app.state.nats = await _connect_nats() if settings.NATS_ENABLED else None app.state.nats = None
if not settings.NATS_ENABLED: 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)") logger.info("NATS выключен (NATS_ENABLED=false)")
try: try:
@ -112,7 +71,7 @@ async def lifespan(app: FastAPI):
finally: finally:
logger.info("Приложение закрывается") logger.info("Приложение закрывается")
if app.state.nats: if app.state.nats:
await app.state.nats.close() await app.state.nats.stop()
if app.state.storage: if app.state.storage:
await app.state.storage.client.close_session() await app.state.storage.client.close_session()