"""Точка входа FastAPI. Каркас из шаблона команды: инфраструктура без доменного кода. Что здесь есть: настройки, подключение S3 (MinIO; `MINIO_SECURE` — https или http), NATS (`src/core/nats.py`: core NATS без JetStream, подключение в фоне при NATS_ENABLED=true независимо от DEBUG, подписка `recognition.result` в queue group `arm-core`), SMB (горячие папки ИС Сканирование; необязателен — при пустом `SMB_HOST` не подключается), middleware (gzip, proxy headers, CORS; при DEBUG=False пакет РАИСА лишь условно импортируется, middleware не подключается), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS. Сообщения `recognition.result` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий приложения и `NoopMapper` (маппинг результата в доменные таблицы — Срез 2). Остановка: фоновые задачи обмена → NATS → S3 → SMB → пул соединений БД; каждый шаг независим — ошибка одного логируется и не отменяет остальные. """ import asyncio import os from collections.abc import Awaitable, Callable from contextlib import asynccontextmanager from datetime import UTC, datetime from functools import partial from http import HTTPStatus import smbclient from fastapi import FastAPI, Request from fastapi.exceptions import HTTPException, StarletteHTTPException from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles 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 API_PREFIX, 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.nats import MessageHandler, NatsConnection from src.db.db import async_session_maker, engine from src.services.recognition_exchange import ExchangeTasks, NoopMapper, handle_message settings = get_settings() VERSION = settings.VERSION UP_TIME = datetime.now(UTC) # --- РАИСА: защищённый API (только вне DEBUG) --------------------------------------------------- protected_middleware_available = False if not settings.DEBUG: try: from raisa_fastapi_protected_api import ( # noqa: F401 # pyright: ignore[reportMissingImports] AuthorizationMiddleware, OpenEndpoint, ProtectedOAuthSettings, ProtectedRolesSettings, ProtectedSettings, SearchType, ) protected_middleware_available = True except ImportError: protected_middleware_available = False def _normalize_root_path(value: str) -> str: if not value: return "" value = value.strip() if value == "/": return "" return "/" + value.strip("/") _APP_ROOT_PATH = _normalize_root_path(settings.ROOT_PATH) async def _connect_s3() -> S3Storage | None: """Клиент S3 с проверкой бакета; при недоступности — ``None`` (приложение всё равно стартует).""" s3_client = Minio( settings.OPENBAO.MINIO.MINIO_URL, access_key=settings.OPENBAO.MINIO.MINIO_ADMIN_LOGIN, secret_key=settings.OPENBAO.MINIO.MINIO_ADMIN_PASSWORD, secure=settings.OPENBAO.MINIO.MINIO_SECURE, cert_check=False, ) try: bucket_exists = await s3_client.bucket_exists(settings.OPENBAO.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), ) await s3_client.close_session() return None logger.info("Бакет проверен. bucket_exists={bucket_exists}", bucket_exists=bucket_exists) return S3Storage(client=s3_client, bucket_name=settings.OPENBAO.MINIO.MINIO_BUCKET) def _result_handler() -> MessageHandler: """Inbox `recognition.result`: `handle_message` с фабрикой сессий приложения и `NoopMapper` (Срез 0).""" return partial(handle_message, session_factory=async_session_maker, mapper=NoopMapper()) async def _subscribe_results(nats: NatsConnection) -> None: """Подписки после первого подключения к NATS: `recognition.result` в queue group `arm-core`.""" await nats.subscribe_results(_result_handler()) def _create_nats() -> NatsConnection | None: """Соединение NATS при NATS_ENABLED=true; при false — ``None`` (healthcheck покажет ``off``).""" if not settings.NATS_ENABLED: logger.info("NATS выключен (NATS_ENABLED=false)") return None return NatsConnection(settings.OPENBAO.NATS, on_connected=_subscribe_results) def _start_exchange(nats: NatsConnection) -> ExchangeTasks: """Фоновые задачи обмена: публикатор outbox и вахтёр сроков (только вместе с NATS, т.е. при NATS_ENABLED=true).""" return ExchangeTasks.start(async_session_maker, nats, settings) def _smb_configured() -> bool: return bool(settings.OPENBAO.SMB.SMB_HOST) async def _register_smb() -> None: """Сессия SMB (шаблон) — только если задан `SMB_HOST`; вызов блокирующий, поэтому в отдельном потоке.""" smb = settings.OPENBAO.SMB if not _smb_configured(): logger.debug("SMB не настроен (OPENBAO__SMB__SMB_HOST пуст): сессия SMB не регистрируется") return try: await asyncio.to_thread(smbclient.register_session, smb.SMB_HOST, smb.SMB_LOGIN, smb.SMB_PASSWORD) except Exception: logger.exception("Ошибка при регистрации сессии SMB") async def _reset_smb() -> None: """Закрывает кэш соединений SMB (блокирующий вызов — в отдельном потоке); без SMB ничего не делает.""" if _smb_configured(): await asyncio.to_thread(smbclient.reset_connection_cache) async def _cleanup_step(name: str, action: Callable[[], Awaitable[object]]) -> None: """Один шаг остановки: исключение логируется и не мешает следующим шагам.""" try: await action() except Exception: logger.exception("Ошибка при остановке приложения. step={step}", step=name) async def _shutdown(app: FastAPI) -> None: """Остановка по шагам: задачи обмена (до NATS: публикатор не пишет в закрытое соединение) → NATS → S3 → SMB → пул соединений БД (последним: задачи обмена и inbox к этому моменту остановлены).""" exchange, nats, storage = app.state.exchange, app.state.nats, app.state.storage if exchange is not None: await _cleanup_step("exchange", exchange.stop) if nats is not None: await _cleanup_step("nats", nats.stop) if storage: await _cleanup_step("s3", storage.client.close_session) await _cleanup_step("smb", _reset_smb) await _cleanup_step("db", engine.dispose) @asynccontextmanager async def lifespan(app: FastAPI): app.state.storage = None app.state.nats = None app.state.exchange = None await _register_smb() try: app.state.storage = await _connect_s3() app.state.nats = _create_nats() if app.state.nats is not None: await app.state.nats.start() # подключение в фоне: старт приложения не ждёт NATS app.state.exchange = _start_exchange(app.state.nats) # публикатор сам ждёт соединения logger.info("Приложение запущено") yield finally: logger.info("Приложение закрывается") await _shutdown(app) app = FastAPI(title="АРМ КА", version=VERSION, root_path=settings.ROOT_PATH, lifespan=lifespan) @app.get("/healthcheck", status_code=200) async def healthcheck(request: Request): """Живость приложения. ``nats``: ``off`` — NATS выключен, ``ok`` — подключён, ``down`` — включён, но нет связи.""" nats_status = "off" if settings.NATS_ENABLED: nc = getattr(request.app.state, "nats", None) nats_status = "ok" if nc is not None and nc.is_connected else "down" return {"uptime": UP_TIME, "version": VERSION, "nats": nats_status} app.include_router(router) register_exception_handlers(app) app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=1) app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"]) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # --- Раздача собранного SPA (web/) --------------------------------------------------------------- os.makedirs("web", exist_ok=True) def sep_static(path: str, stat_dir: str) -> str: sep_path = path.split("/") index = sep_path.index(stat_dir) return "/".join(sep_path[index:]) API_STATIC_PATH = API_PREFIX.strip("/") class SPAStaticFiles(StaticFiles): async def get_response(self, path: str, scope): # Несуществующий маршрут API — честный 404 (в т.ч. для POST), а не index.html SPA и не 405 от StaticFiles: # иначе dev-маршруты, которых при выключенных флагах нет, выглядели бы как «существующие» if path == API_STATIC_PATH or path.startswith(API_STATIC_PATH + "/"): raise HTTPException(status_code=HTTPStatus.NOT_FOUND) try: return await super().get_response(path, scope) except (HTTPException, StarletteHTTPException) as ex: if ex.status_code == HTTPStatus.NOT_FOUND: if "images" in path: return await super().get_response(sep_static(path, "images"), scope) if "static" in path: return await super().get_response(sep_static(path, "static"), scope) return await super().get_response("index.html", scope) raise app.mount("/", SPAStaticFiles(directory="web", html=True), name="web")