Ali Zarinkolah
3bced65926
Why: - Wires the ADR-0008 error envelope, per-request correlation id, and the /v1/files routes into the app. Changes: - Extend AppResources/lifespan with the ingestion CapacityLimiter and ObjectStorage adapter. Impact: - /v1 now exposes routes for the first time.
79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
from collections.abc import AsyncIterator, Callable
|
|
from contextlib import AbstractAsyncContextManager, asynccontextmanager
|
|
|
|
import structlog
|
|
from anyio import CapacityLimiter, to_thread
|
|
from fastapi import FastAPI
|
|
|
|
from src.application.ingestion import get_encoder
|
|
from src.bootstrap.dependencies import AppResources
|
|
from src.config import Settings
|
|
from src.infrastructure.minio.client import create_client as create_minio_client
|
|
from src.infrastructure.minio.storage import MinioObjectStorage
|
|
from src.infrastructure.observability.logging import configure_logging
|
|
from src.infrastructure.postgres.database import create_engine, create_sessionmaker
|
|
from src.infrastructure.qdrant.client import create_client as create_qdrant_client
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
def create_lifespan(
|
|
settings: Settings | None = None,
|
|
) -> Callable[[FastAPI], AbstractAsyncContextManager[None, bool | None]]:
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
|
|
resolved_settings = settings or Settings()
|
|
configure_logging(resolved_settings.logging)
|
|
|
|
# tiktoken fetches its vocabulary over the network on first use, so warm
|
|
# it here: a missing vocabulary should fail the process at boot, not the
|
|
# first upload. Blocking, hence the thread.
|
|
await to_thread.run_sync(get_encoder, resolved_settings.chunking.encoding_name)
|
|
logger.info(
|
|
"lifespan.tokenizer.loaded",
|
|
encoding=resolved_settings.chunking.encoding_name,
|
|
)
|
|
|
|
db_engine = create_engine(resolved_settings.postgres)
|
|
db_sessionmaker = create_sessionmaker(db_engine)
|
|
logger.info("lifespan.postgres.engine.created")
|
|
|
|
minio_client = create_minio_client(resolved_settings.minio)
|
|
logger.info("lifespan.minio.client.created")
|
|
|
|
qdrant_client = create_qdrant_client(resolved_settings.qdrant)
|
|
logger.info("lifespan.qdrant.client.created")
|
|
|
|
# Bounds threads spent on blocking ingestion work (parsing, chunking,
|
|
# hashing, the sync minio SDK) so it cannot exhaust Starlette's own
|
|
# thread pool (ADR-0017).
|
|
ingestion_limiter = CapacityLimiter(resolved_settings.ingestion.thread_pool_size)
|
|
object_storage = MinioObjectStorage(
|
|
minio_client, bucket=resolved_settings.minio.bucket, limiter=ingestion_limiter
|
|
)
|
|
|
|
app.state.resources = AppResources(
|
|
settings=resolved_settings,
|
|
db_engine=db_engine,
|
|
db_sessionmaker=db_sessionmaker,
|
|
minio_client=minio_client,
|
|
qdrant_client=qdrant_client,
|
|
object_storage=object_storage,
|
|
ingestion_limiter=ingestion_limiter,
|
|
)
|
|
|
|
try:
|
|
yield
|
|
finally:
|
|
try:
|
|
await db_engine.dispose()
|
|
except Exception:
|
|
logger.exception("lifespan.postgres.dispose.failed")
|
|
|
|
try:
|
|
await qdrant_client.close()
|
|
except Exception:
|
|
logger.exception("lifespan.qdrant.close.failed")
|
|
|
|
return lifespan
|