feat: запуск воркера в lifespan при NATS_ENABLED и поле nats = ok | down | off в /healthcheck
Подписка — через on_connected NatsConnection; остановка: воркер (дорабатывает очередь) → NATS → S3. Клиент S3 остаётся и при неудачной проверке бакета, отсутствие бакета — warning (T2.1). Тест lifespan и healthcheck с фейками (T8.7); формулировки nats в README, .env.example и spec приведены к R-17. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
e1231dcdb4
commit
a7862c1843
@ -48,8 +48,9 @@ Stateless-сервис распознавания табличных файло
|
||||
```
|
||||
api/ бэкенд (все команды разработки — отсюда или через make из корня)
|
||||
run_server.py точка входа платформы (uvicorn; SERVER__HOST, SERVER__PORT)
|
||||
src/app.py FastAPI: lifespan (подключение к MinIO и NATS), GET /healthcheck, middleware, обработчики ошибок
|
||||
src/core/ config.py (настройки), clients.py (S3), logger.py, exception_handlers.py
|
||||
src/app.py FastAPI: lifespan (MinIO, NATS, воркер), GET /healthcheck, middleware, обработчики ошибок
|
||||
src/core/ config.py (настройки), clients.py (S3), nats.py (соединение NATS), worker.py (воркер
|
||||
распознавания), logger.py, exception_handlers.py
|
||||
src/parser/ интерфейс Parser и реализации (добавляется задачей 4.1)
|
||||
src/contract/ pydantic-модели контракта v1.0 (добавляется задачей 3.1)
|
||||
contracts/ копии JSON-схем контракта (добавляется задачей 3.1)
|
||||
@ -108,8 +109,8 @@ cd api && .venv/bin/python run_server.py
|
||||
curl -s localhost:8000/healthcheck
|
||||
```
|
||||
|
||||
В ответе — `uptime`, `version` и (после добавления воркера, задача 4.3) `nats`: `ok` — подключён, `down` — включён
|
||||
(`NATS_ENABLED=true`), но связи нет, `off` — выключен.
|
||||
В ответе — `uptime`, `version` и `nats`: `ok` — подключён, `down` — включён (`NATS_ENABLED=true`), но связи нет,
|
||||
`off` — выключен (`NATS_ENABLED=false`, воркер не запущен).
|
||||
|
||||
Если NATS или MinIO недоступны при старте, приложение всё равно поднимается: ошибка логируется, соединение с NATS
|
||||
восстанавливается автоматически (бесконечное автопереподключение).
|
||||
|
||||
@ -27,7 +27,7 @@ ROOT_PATH=/
|
||||
VERSION=1.0.0
|
||||
# `service.name` конвертов контракта и имя queue group NATS; менять не нужно, пока контракт v1.x
|
||||
SERVICE_NAME=tabular-parser
|
||||
# false — не подключаться к NATS (воркер не стартует; /healthcheck отдаёт nats = down)
|
||||
# false — не подключаться к NATS (воркер не стартует; /healthcheck отдаёт nats = off)
|
||||
NATS_ENABLED=true
|
||||
# Число заданий, обрабатываемых одновременно
|
||||
WORKER_CONCURRENCY=1
|
||||
|
||||
@ -1,15 +1,18 @@
|
||||
"""Точка входа FastAPI сервиса tabular-parser.
|
||||
|
||||
Что здесь есть: настройки, подключение 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.
|
||||
аутентификации, фоновое подключение — старт не ждёт NATS), воркер `src/core/worker.py` (подписка на
|
||||
`recognition.tabular.request` после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck`
|
||||
(`nats = ok | down | off`, R-17), обработчики исключений. Сервис без БД и фронта (DECISION-4).
|
||||
|
||||
При `NATS_ENABLED=false` ни NATS, ни воркер не запускаются. Остановка: воркер перестаёт принимать запросы и
|
||||
дорабатывает очередь (NATS ещё нужен для публикации результатов), затем закрывается NATS (drain), затем сессия S3.
|
||||
"""
|
||||
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi import FastAPI, Request
|
||||
from loguru import logger
|
||||
from miniopy_async import Minio
|
||||
from starlette.middleware.gzip import GZipMiddleware
|
||||
@ -21,6 +24,8 @@ 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
|
||||
from src.core.worker import STOP_TIMEOUT, RecognitionWorker
|
||||
from src.parser import get_parser
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
@ -30,8 +35,12 @@ UP_TIME = datetime.now(UTC)
|
||||
setup_logger(level="DEBUG" if settings.DEBUG else "INFO")
|
||||
|
||||
|
||||
async def _connect_s3() -> S3Storage | None:
|
||||
"""Клиент S3 с проверкой бакета; при недоступности — ``None`` (приложение всё равно стартует)."""
|
||||
async def _connect_s3() -> S3Storage:
|
||||
"""Клиент S3 с проверкой бакета.
|
||||
|
||||
Проверка — только диагностика: при недоступном MinIO или отсутствующем бакете ошибка логируется, а клиент
|
||||
остаётся (соединения открываются на каждый запрос) — когда MinIO поднимется, воркер прочитает файл без рестарта.
|
||||
"""
|
||||
minio = settings.OPENBAO.MINIO
|
||||
s3_client = Minio(
|
||||
minio.MINIO_URL,
|
||||
@ -48,40 +57,79 @@ async def _connect_s3() -> S3Storage | None:
|
||||
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)
|
||||
else:
|
||||
if bucket_exists:
|
||||
logger.info("Бакет проверен. bucket={bucket}", bucket=minio.MINIO_BUCKET)
|
||||
else:
|
||||
logger.warning(
|
||||
"Бакет не найден: чтение исходников и запись результатов будут падать. error_category=s3_error, "
|
||||
"bucket={bucket}",
|
||||
bucket=minio.MINIO_BUCKET,
|
||||
)
|
||||
return S3Storage(client=s3_client, bucket_name=minio.MINIO_BUCKET)
|
||||
|
||||
|
||||
def _build_worker(storage: S3Storage) -> tuple[NatsConnection, RecognitionWorker]:
|
||||
"""Соединение NATS и воркер; подписка на запросы — в `on_connected` (после первого подключения)."""
|
||||
|
||||
async def subscribe(conn: NatsConnection) -> None:
|
||||
await conn.subscribe_requests(worker.on_request)
|
||||
|
||||
nats = NatsConnection(settings.OPENBAO.NATS, on_connected=subscribe, publish_retries=settings.PUBLISH_RETRIES)
|
||||
worker = RecognitionWorker(
|
||||
storage=storage,
|
||||
publisher=nats,
|
||||
parser=get_parser(),
|
||||
service_name=settings.SERVICE_NAME,
|
||||
service_version=VERSION,
|
||||
concurrency=settings.WORKER_CONCURRENCY,
|
||||
done_cache_size=settings.DONE_CACHE_SIZE,
|
||||
)
|
||||
return nats, worker
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
app.state.storage = await _connect_s3()
|
||||
app.state.nats = None
|
||||
storage = await _connect_s3()
|
||||
nats: NatsConnection | None = None
|
||||
worker: RecognitionWorker | None = None
|
||||
if settings.NATS_ENABLED:
|
||||
nats, worker = _build_worker(storage)
|
||||
worker.start()
|
||||
# start() не ждёт NATS: при недоступном сервере подключение продолжается в фоне (design D-6)
|
||||
app.state.nats = NatsConnection(settings.OPENBAO.NATS, publish_retries=settings.PUBLISH_RETRIES)
|
||||
await app.state.nats.start()
|
||||
await nats.start()
|
||||
else:
|
||||
logger.info("NATS выключен (NATS_ENABLED=false)")
|
||||
logger.info("NATS выключен (NATS_ENABLED=false): воркер не запускается")
|
||||
app.state.storage = storage
|
||||
app.state.nats = nats
|
||||
app.state.worker = worker
|
||||
|
||||
try:
|
||||
logger.info("Приложение запущено")
|
||||
yield
|
||||
finally:
|
||||
logger.info("Приложение закрывается")
|
||||
if app.state.nats:
|
||||
await app.state.nats.stop()
|
||||
if app.state.storage:
|
||||
await app.state.storage.client.close_session()
|
||||
if worker is not None:
|
||||
await worker.stop(STOP_TIMEOUT) # результаты ещё публикуются: NATS закрывается после воркера
|
||||
if nats is not None:
|
||||
await nats.stop()
|
||||
await storage.client.close_session()
|
||||
|
||||
|
||||
app = FastAPI(title="tabular-parser", version=VERSION, root_path=settings.ROOT_PATH, lifespan=lifespan)
|
||||
|
||||
|
||||
def _nats_state(application: FastAPI) -> str:
|
||||
"""`off` — NATS выключен флагом (R-17); `ok` — соединение есть; `down` — включён, но связи нет."""
|
||||
if not settings.NATS_ENABLED:
|
||||
return "off"
|
||||
nats = getattr(application.state, "nats", None)
|
||||
return "ok" if nats is not None and nats.is_connected else "down"
|
||||
|
||||
|
||||
@app.get("/healthcheck", status_code=200)
|
||||
async def healthcheck():
|
||||
return {"uptime": UP_TIME, "version": VERSION}
|
||||
async def healthcheck(request: Request):
|
||||
return {"uptime": UP_TIME, "version": VERSION, "nats": _nats_state(request.app)}
|
||||
|
||||
|
||||
app.include_router(router)
|
||||
|
||||
@ -1,19 +0,0 @@
|
||||
"""Обработчики сообщений NATS.
|
||||
|
||||
Каркас: доменный обработчик удалён. Consumer `recognition.result` (дедуп по recognition_job.result_msg_id,
|
||||
маппинг JSON из S3 в document/osv_entry/financial_statement_*) реализуется в Срезе 0–2 по
|
||||
docs/context/03-architecture.md. Для референса — reference-old-arm/api/src/core/nats_callback.py.
|
||||
"""
|
||||
|
||||
from loguru import logger
|
||||
from nats.aio.client import Client as NATS
|
||||
|
||||
from src.core.clients import S3Storage
|
||||
|
||||
|
||||
def build_result_handler(nc: NATS, storage: S3Storage):
|
||||
async def handler(msg):
|
||||
logger.info("NATS message received. subject={subject}, size={size}", subject=msg.subject, size=len(msg.data))
|
||||
await msg.ack()
|
||||
|
||||
return handler
|
||||
179
api/tests/unit/test_app.py
Normal file
179
api/tests/unit/test_app.py
Normal file
@ -0,0 +1,179 @@
|
||||
"""Unit-тесты lifespan и `/healthcheck` приложения без NATS и MinIO (design D-6, R-17; carry-over T8.7).
|
||||
|
||||
`NatsConnection`, `_connect_s3` и `RecognitionWorker` модуля `src.app` подменяются фейками: проверяется, что при
|
||||
`NATS_ENABLED=true` воркер запускается и подписывается через `on_connected`, порядок остановки
|
||||
(воркер → NATS → S3) и значения `nats = ok | down | off`.
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import json
|
||||
from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from src.contract import NATS_MSG_ID_HEADER, RESULT_SUBJECT
|
||||
from src.core.worker import RecognitionWorker
|
||||
|
||||
EXAMPLES_DIR = Path(__file__).resolve().parents[2] / "contracts" / "examples"
|
||||
REQUEST = json.loads((EXAMPLES_DIR / "request-tabular.json").read_text(encoding="utf-8"))
|
||||
JOB_ID = REQUEST["job_id"]
|
||||
app_module = importlib.import_module("src.app") # модуль, а не объект `app`, который реэкспортирует пакет `src`
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def running_app(module: Any) -> AsyncIterator[httpx.AsyncClient]:
|
||||
"""Lifespan приложения + HTTP-клиент поверх ASGI (без сети)."""
|
||||
application = module.app
|
||||
async with application.router.lifespan_context(application):
|
||||
transport = httpx.ASGITransport(app=application)
|
||||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||||
yield client
|
||||
|
||||
|
||||
class FakeStorage:
|
||||
def __init__(self, events: list[str]) -> None:
|
||||
self.events = events
|
||||
self.client = SimpleNamespace(close_session=self._close)
|
||||
self.objects = {(REQUEST["s3"]["bucket"], REQUEST["s3"]["key"]): b"xlsx"}
|
||||
|
||||
async def _close(self) -> None:
|
||||
self.events.append("s3.close")
|
||||
|
||||
async def get_bytes(self, bucket: str, key: str) -> bytes:
|
||||
return self.objects[(bucket, key)]
|
||||
|
||||
async def put_json(self, bucket: str, key: str, obj: Any) -> None:
|
||||
raise AssertionError("NullParser не пишет файл результата")
|
||||
|
||||
|
||||
class FakeNatsConnection:
|
||||
"""Повторяет интерфейс `NatsConnection`, нужный lifespan и воркеру."""
|
||||
|
||||
instances: list["FakeNatsConnection"] = []
|
||||
connect_on_start = True
|
||||
events: list[str] = [] # общий журнал порядка запуска и остановки (задаётся фикстурой)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
settings: Any,
|
||||
*,
|
||||
on_connected: Callable[["FakeNatsConnection"], Awaitable[None]] | None = None,
|
||||
publish_retries: int = 3,
|
||||
) -> None:
|
||||
self.settings = settings
|
||||
self.on_connected = on_connected
|
||||
self.publish_retries = publish_retries
|
||||
self.is_connected = False
|
||||
self.handlers: list[tuple[str, Callable[[Any], Awaitable[None]]]] = []
|
||||
self.published: list[tuple[str, dict[str, Any], str]] = []
|
||||
FakeNatsConnection.instances.append(self)
|
||||
|
||||
async def start(self) -> None:
|
||||
self.events.append("nats.start")
|
||||
if self.connect_on_start:
|
||||
self.is_connected = True
|
||||
if self.on_connected is not None:
|
||||
await self.on_connected(self)
|
||||
|
||||
async def stop(self) -> None:
|
||||
self.events.append("nats.stop")
|
||||
self.is_connected = False
|
||||
|
||||
async def subscribe_requests(self, cb: Callable[[Any], Awaitable[None]]) -> None:
|
||||
self.handlers.append(("recognition.tabular.request", cb))
|
||||
|
||||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
|
||||
self.published.append((subject, json.loads(payload), msg_id))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def events() -> list[str]:
|
||||
return []
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def patched_app(monkeypatch: pytest.MonkeyPatch, events: list[str]) -> Iterator[Any]:
|
||||
"""`src.app` с фейковыми S3, NATS и записывающим порядок остановки воркером."""
|
||||
FakeNatsConnection.instances = []
|
||||
FakeNatsConnection.connect_on_start = True
|
||||
FakeNatsConnection.events = events
|
||||
storage = FakeStorage(events)
|
||||
|
||||
async def fake_connect_s3() -> FakeStorage:
|
||||
return storage
|
||||
|
||||
class RecordingWorker(RecognitionWorker):
|
||||
async def stop(self, timeout: float = 10.0) -> None:
|
||||
events.append("worker.stop")
|
||||
await super().stop(timeout)
|
||||
|
||||
monkeypatch.setattr(app_module, "_connect_s3", fake_connect_s3)
|
||||
monkeypatch.setattr(app_module, "NatsConnection", FakeNatsConnection)
|
||||
monkeypatch.setattr(app_module, "RecognitionWorker", RecordingWorker)
|
||||
yield app_module
|
||||
|
||||
|
||||
async def test_nats_disabled_starts_nothing_and_healthcheck_reports_off(
|
||||
patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str]
|
||||
) -> None:
|
||||
monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", False)
|
||||
|
||||
async with running_app(patched_app) as client:
|
||||
response = await client.get("/healthcheck")
|
||||
assert patched_app.app.state.worker is None
|
||||
assert patched_app.app.state.nats is None
|
||||
|
||||
assert response.status_code == 200 # noqa: PLR2004 - HTTP 200
|
||||
body = response.json()
|
||||
assert set(body) == {"uptime", "version", "nats"}
|
||||
assert body["nats"] == "off"
|
||||
assert body["version"] == patched_app.settings.VERSION
|
||||
assert FakeNatsConnection.instances == []
|
||||
assert events == ["s3.close"]
|
||||
|
||||
|
||||
async def test_nats_enabled_subscribes_worker_and_reports_ok(
|
||||
patched_app: Any, monkeypatch: pytest.MonkeyPatch, events: list[str]
|
||||
) -> None:
|
||||
monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", True)
|
||||
|
||||
async with running_app(patched_app) as client:
|
||||
assert (await client.get("/healthcheck")).json()["nats"] == "ok"
|
||||
(connection,) = FakeNatsConnection.instances
|
||||
assert connection.publish_retries == patched_app.settings.PUBLISH_RETRIES
|
||||
assert connection.handlers, "воркер должен подписаться через on_connected"
|
||||
worker = patched_app.app.state.worker
|
||||
assert isinstance(worker, RecognitionWorker)
|
||||
|
||||
# Запрос, доставленный подписке, проходит через воркер: ack и результат NOT_IMPLEMENTED в recognition.result
|
||||
_subject, handler = connection.handlers[0]
|
||||
message = SimpleNamespace(
|
||||
data=json.dumps(REQUEST).encode(), headers={NATS_MSG_ID_HEADER: f"{JOB_ID}:1"}, subject=_subject
|
||||
)
|
||||
await handler(message)
|
||||
await worker.join()
|
||||
|
||||
assert [(subject, msg_id) for subject, _body, msg_id in connection.published] == [
|
||||
(RESULT_SUBJECT, f"ack:{JOB_ID}:1"),
|
||||
(RESULT_SUBJECT, f"result:{JOB_ID}:1"),
|
||||
]
|
||||
assert connection.published[1][1]["error"]["code"] == "NOT_IMPLEMENTED"
|
||||
assert connection.published[1][1]["service"]["name"] == "tabular-parser"
|
||||
# Остановка: воркер дорабатывает очередь, пока NATS ещё жив, затем NATS, затем S3
|
||||
assert events == ["nats.start", "worker.stop", "nats.stop", "s3.close"]
|
||||
|
||||
|
||||
async def test_nats_enabled_but_not_connected_reports_down(patched_app: Any, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(patched_app.settings, "NATS_ENABLED", True)
|
||||
FakeNatsConnection.connect_on_start = False
|
||||
|
||||
async with running_app(patched_app) as client:
|
||||
assert (await client.get("/healthcheck")).json()["nats"] == "down"
|
||||
(connection,) = FakeNatsConnection.instances
|
||||
assert connection.handlers == [] # подписка — только после подключения
|
||||
assert patched_app.app.state.worker is not None
|
||||
@ -80,7 +80,8 @@ NATS и S3; воркер MUST NOT зависеть от конкретной р
|
||||
### Requirement: Соединение с NATS по правилам шаблона, без JetStream
|
||||
При пустых `NATS_SEED`/`NATS_JWT` сервис SHALL подключаться без учётных данных, иначе — с nkeys/JWT. Сервис MUST NOT
|
||||
требовать stream'ов или consumer'ов JetStream; при разрыве соединения SHALL переподключаться автоматически и восстанавливать
|
||||
подписку; `/healthcheck` SHALL отражать состояние (`nats = ok | down`).
|
||||
подписку; `/healthcheck` SHALL отражать состояние (`nats = ok | down`; `off` — NATS выключен флагом
|
||||
`NATS_ENABLED=false`, R-17).
|
||||
|
||||
#### Scenario: NATS перезапущен
|
||||
- **WHEN** контейнер NATS перезапущен во время работы сервиса
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user