ee272b2b92
Функция осталась от шаблона и нигде не вызывается (grep по api/, включая тесты). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
115 lines
5.8 KiB
Python
115 lines
5.8 KiB
Python
"""Точка входа FastAPI сервиса scan-gateway (адаптер ИС Сканирование). Без БД и без фронта.
|
||
|
||
- `create_app(settings)` строит горячую папку (`create_hotfolder`, без обращения к ФС/шаре) и `HealthProvider`;
|
||
`GET /healthcheck` отдаёт `uptime`, `version`, `nats` (`ok`/`down`/`disabled`), `hotfolder` (`ok`/`down`).
|
||
- lifespan: регистрация SMB-сессии (только `SCAN_HOTFOLDER_MODE=smb`), S3 (`S3Storage`), NATS (`NatsGateway`), воркер
|
||
(`Worker` с `NullNormalizer`). Подключение к NATS блокирует до доступности сервера, поэтому идёт фоновой задачей:
|
||
HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после подключения. Сбой этой
|
||
задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`, HTTP-сервер жив.
|
||
Остановка: отмена подключения, остановка воркера (задания в работе отменяются), закрытие NATS и S3.
|
||
- Объекты lifespan доступны в `app.state`: `worker`, `nats`, `storage`, `health` (тесты, задачи 13–14).
|
||
- `NATS_ENABLED=false` — подключения нет (`nats = disabled`), воркер запросов не получает (dev без NATS).
|
||
"""
|
||
|
||
import asyncio
|
||
from contextlib import asynccontextmanager, suppress
|
||
from datetime import UTC, datetime
|
||
|
||
import smbclient
|
||
from fastapi import FastAPI
|
||
from loguru import logger
|
||
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 Settings, get_settings
|
||
from src.core.exception_handlers import register_exception_handlers
|
||
from src.core.health import HealthProvider
|
||
from src.core.nats import NatsGateway
|
||
from src.core.worker import Worker
|
||
from src.gateway import create_hotfolder
|
||
from src.normalizer import NullNormalizer
|
||
|
||
|
||
async def run_nats(gateway: NatsGateway, worker: Worker) -> None:
|
||
"""Фоновая задача: подключение к NATS (ждёт сервер сколь угодно долго) и подписка воркера на запросы.
|
||
|
||
Ошибка не роняет приложение: лог error с трассировкой; если соединение установлено, а подписка не удалась,
|
||
соединение закрывается, чтобы `/healthcheck.nats = down` честно показывал, что запросы не принимаются.
|
||
"""
|
||
try:
|
||
if not await gateway.connect():
|
||
return
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception:
|
||
logger.exception("NATS: подключение не выполнено, запросы не принимаются (nats=down) до перезапуска")
|
||
return
|
||
try:
|
||
await gateway.subscribe_requests(worker.on_request)
|
||
except asyncio.CancelledError:
|
||
raise
|
||
except Exception:
|
||
logger.exception("NATS: подписка на запросы не оформлена, соединение закрывается (nats=down)")
|
||
await gateway.close()
|
||
|
||
|
||
def create_app(settings: Settings | None = None) -> FastAPI:
|
||
settings = settings or get_settings()
|
||
version = settings.VERSION
|
||
up_time = datetime.now(UTC)
|
||
hotfolder = create_hotfolder(settings)
|
||
health = HealthProvider(nats_enabled=settings.NATS_ENABLED, hotfolder=hotfolder)
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
smb_session = False
|
||
if settings.SCAN_HOTFOLDER_MODE == "smb":
|
||
smb = settings.OPENBAO.SMB
|
||
try:
|
||
smbclient.register_session(smb.SMB_HOST, smb.SMB_LOGIN, smb.SMB_PASSWORD)
|
||
smb_session = True
|
||
except Exception:
|
||
# учётные данные передаются в каждый вызов smbclient (SmbHotFolder), сессия восстановится при
|
||
# доступной шаре; пока шара недоступна, /healthcheck.hotfolder = down
|
||
logger.exception("Ошибка при регистрации сессии SMB")
|
||
storage = S3Storage.from_settings(settings.OPENBAO.MINIO)
|
||
gateway = NatsGateway(settings)
|
||
worker = Worker(settings, storage=storage, publisher=gateway, hotfolder=hotfolder, normalizer=NullNormalizer())
|
||
health.nats = gateway
|
||
app.state.storage = storage
|
||
app.state.nats = gateway
|
||
app.state.worker = worker
|
||
worker.start()
|
||
nats_task = asyncio.create_task(run_nats(gateway, worker), name="nats-connect")
|
||
try:
|
||
logger.info("Приложение запущено")
|
||
yield
|
||
finally:
|
||
logger.info("Приложение закрывается")
|
||
nats_task.cancel()
|
||
with suppress(asyncio.CancelledError):
|
||
await nats_task
|
||
await worker.stop()
|
||
await gateway.close()
|
||
await storage.close()
|
||
if smb_session:
|
||
smbclient.reset_connection_cache()
|
||
|
||
app = FastAPI(title="scan-gateway", version=version, root_path=settings.ROOT_PATH, lifespan=lifespan)
|
||
app.state.health = health
|
||
|
||
@app.get("/healthcheck", status_code=200)
|
||
async def healthcheck():
|
||
return {"uptime": up_time, "version": version, **await health.snapshot()}
|
||
|
||
app.include_router(router)
|
||
register_exception_handlers(app)
|
||
app.add_middleware(GZipMiddleware, minimum_size=1000, compresslevel=1)
|
||
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts=["*"])
|
||
return app
|
||
|
||
|
||
app = create_app()
|