"""Точка входа FastAPI. Каркас из шаблона команды: инфраструктура без доменного кода. Что здесь есть: настройки, подключение S3 (MinIO), NATS (nkeys JWT; пустые NATS_SEED/NATS_JWT — без учётных данных; подключение при NATS_ENABLED=true независимо от DEBUG), SMB (горячие папки ИС Сканирование), middleware (gzip, proxy headers, CORS, РАИСА при DEBUG=False), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Недоступность S3 или NATS при старте не роняет приложение: ошибка логируется, healthcheck показывает состояние. Что добавляет Срез 0: роутеры в src/api/router.py, outbox-паблишер и consumer recognition.result (docs/context/03-architecture.md), зависимость require(permission). """ import base64 import os from contextlib import asynccontextmanager from datetime import UTC, datetime from http import HTTPStatus from typing import Any import nkeys 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 nats.aio.client import Client as NATS 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 NATSSettings, get_settings from src.core.exception_handlers import register_exception_handlers settings = get_settings() VERSION = settings.VERSION UP_TIME = datetime.now(UTC) # --- NATS: аутентификация по nkeys/JWT --------------------------------------------------------- def _build_nats_auth(nats_settings: NATSSettings) -> dict[str, Any]: """Параметры аутентификации для ``NATS.connect``. Пустые NATS_SEED и NATS_JWT — подключение без учётных данных (docker compose, тесты): ``{}``. Заданы оба — колбэки nkeys/JWT, как в шаблоне. Задан только один — ошибка конфигурации. Подключение переедет в core/nats.py (задача 22). """ seed = nats_settings.NATS_SEED jwt = nats_settings.NATS_JWT if not seed and not jwt: return {} if not seed or not jwt: raise RuntimeError("Неполные учётные данные NATS: NATS_SEED и NATS_JWT задаются вместе или не задаются вовсе") keypair = nkeys.from_seed(seed.encode("utf-8")) def user_jwt_cb() -> bytes: return jwt.encode("utf-8") def signature_cb(nonce: str | bytes) -> bytes: if isinstance(nonce, str): nonce = nonce.encode("utf-8") return base64.b64encode(keypair.sign(nonce)) return {"user_jwt_cb": user_jwt_cb, "signature_cb": signature_cb} # --- РАИСА: защищённый 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, 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) async def _connect_nats() -> NATS | None: """Подключение к NATS при NATS_ENABLED=true; при ошибке — ``None`` (healthcheck покажет ``down``).""" if not settings.NATS_ENABLED: logger.info("NATS выключен (NATS_ENABLED=false)") return None nc = NATS() try: await nc.connect( settings.OPENBAO.NATS.NATS_URL, max_reconnect_attempts=-1, reconnect_time_wait=2, **_build_nats_auth(settings.OPENBAO.NATS), ) 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 @asynccontextmanager async def lifespan(app: FastAPI): app.state.nats_subs = [] app.state.storage = None app.state.nats = None try: smbclient.register_session( settings.OPENBAO.SMB.SMB_HOST, settings.OPENBAO.SMB.SMB_LOGIN, settings.OPENBAO.SMB.SMB_PASSWORD ) except Exception: logger.exception("Ошибка при регистрации сессии SMB") try: app.state.storage = await _connect_s3() app.state.nats = await _connect_nats() # Подписки (recognition.result) и outbox-паблишер — Срез 0, см. docs/context/03-architecture.md logger.info("Приложение запущено") yield finally: logger.info("Приложение закрывается") smbclient.reset_connection_cache() for sub in app.state.nats_subs: await sub.unsubscribe() if app.state.nats: await app.state.nats.close() if app.state.storage: await app.state.storage.client.close_session() 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")