From 3bced65926f4cfd57635f3b6f02221e3839d30f0 Mon Sep 17 00:00:00 2001 From: Ali Zarinkolah Date: Wed, 19 Aug 2026 15:00:39 +0330 Subject: [PATCH] feat(api): add POST/GET /v1/files with auth, error envelope, and request-id middleware 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. --- src/api/dependencies/auth.py | 38 ++++++++++++++ src/api/errors.py | 97 +++++++++++++++++++++++++++++++++++ src/api/middleware.py | 38 ++++++++++++++ src/api/router.py | 3 ++ src/api/routers/files.py | 74 ++++++++++++++++++++++++++ src/api/schemas/errors.py | 16 ++++++ src/api/schemas/files.py | 50 ++++++++++++++++++ src/bootstrap/dependencies.py | 23 +++++++++ src/bootstrap/lifespan.py | 13 ++++- src/main.py | 4 ++ 10 files changed, 355 insertions(+), 1 deletion(-) create mode 100644 src/api/dependencies/auth.py create mode 100644 src/api/errors.py create mode 100644 src/api/middleware.py create mode 100644 src/api/routers/files.py create mode 100644 src/api/schemas/errors.py create mode 100644 src/api/schemas/files.py diff --git a/src/api/dependencies/auth.py b/src/api/dependencies/auth.py new file mode 100644 index 0000000..c29ae96 --- /dev/null +++ b/src/api/dependencies/auth.py @@ -0,0 +1,38 @@ +"""Auth dependencies (ADR-0008): resolve `AuthContext` from a bearer token, +then gate routes on scope. +""" + +from collections.abc import Awaitable, Callable +from typing import Annotated + +from fastapi import Depends +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.auth.context import AuthContext +from src.application.auth.errors import InvalidApiKeyError, MissingScopeError +from src.application.auth.service import resolve_auth_context +from src.bootstrap.dependencies import get_sessionmaker + +_bearer_scheme = HTTPBearer(auto_error=False) + + +async def get_auth_context( + credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer_scheme)], + sessionmaker: Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)], +) -> AuthContext: + if credentials is None: + raise InvalidApiKeyError("missing Authorization header") + return await resolve_auth_context(sessionmaker, credentials.credentials) + + +AuthContextDep = Annotated[AuthContext, Depends(get_auth_context)] + + +def require_scope(scope: str) -> Callable[[AuthContext], Awaitable[AuthContext]]: + async def _dependency(auth: AuthContextDep) -> AuthContext: + if not auth.has_scope(scope): + raise MissingScopeError(f"missing required scope '{scope}'") + return auth + + return _dependency diff --git a/src/api/errors.py b/src/api/errors.py new file mode 100644 index 0000000..e235ce8 --- /dev/null +++ b/src/api/errors.py @@ -0,0 +1,97 @@ +"""Maps application exceptions to the ADR-0008 error envelope. + +This is the single place that knows the exception-type -> status-code +mapping; application/infrastructure code never imports FastAPI or raises +`HTTPException` (ADR-0015). +""" + +import structlog +from fastapi import FastAPI, Request, status +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from starlette.exceptions import HTTPException as StarletteHTTPException + +from src.application.auth.errors import ( + InvalidApiKeyError, + MissingScopeError, + TenantInactiveError, +) +from src.application.files.errors import FileTooLargeError, InvalidUploadError +from src.application.ingestion.errors import DocumentParseError, UnsupportedSourceTypeError + +logger = structlog.get_logger(__name__) + +# (exception type, status code, stable error code) +_MAPPING: tuple[tuple[type[Exception], int, str], ...] = ( + (InvalidApiKeyError, status.HTTP_401_UNAUTHORIZED, "invalid_api_key"), + (TenantInactiveError, status.HTTP_401_UNAUTHORIZED, "tenant_not_found"), + (MissingScopeError, status.HTTP_403_FORBIDDEN, "missing_scope"), + (InvalidUploadError, status.HTTP_400_BAD_REQUEST, "validation_error"), + (DocumentParseError, status.HTTP_400_BAD_REQUEST, "validation_error"), + (UnsupportedSourceTypeError, status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, "unsupported_media_type"), + (FileTooLargeError, status.HTTP_413_CONTENT_TOO_LARGE, "payload_too_large"), +) + + +def _request_id(request: Request) -> str | None: + return getattr(request.state, "request_id", None) + + +def _envelope( + code: str, message: str, request_id: str | None, details: dict[str, object] | None = None +) -> dict[str, object]: + return { + "error": { + "code": code, + "message": message, + "details": details or {}, + "request_id": request_id, + } + } + + +def register_exception_handlers(app: FastAPI) -> None: + for exc_type, status_code, error_code in _MAPPING: + + def _handler( + request: Request, + exc: Exception, + status_code: int = status_code, + error_code: str = error_code, + ) -> JSONResponse: + return JSONResponse( + status_code=status_code, + content=_envelope(error_code, str(exc), _request_id(request)), + ) + + app.add_exception_handler(exc_type, _handler) + + @app.exception_handler(RequestValidationError) + def _validation_handler(request: Request, exc: RequestValidationError) -> JSONResponse: + return JSONResponse( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + content=_envelope( + "validation_error", + "request validation failed", + _request_id(request), + details={"errors": exc.errors()}, + ), + ) + + @app.exception_handler(StarletteHTTPException) + def _http_exception_handler(request: Request, exc: StarletteHTTPException) -> JSONResponse: + code = "not_found" if exc.status_code == status.HTTP_404_NOT_FOUND else "http_error" + return JSONResponse( + status_code=exc.status_code, + content=_envelope(code, str(exc.detail), _request_id(request)), + ) + + @app.exception_handler(Exception) + def _unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse: + logger.exception("api.unhandled_exception", path=request.url.path) + return JSONResponse( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + content=_envelope( + "internal_error", "an unexpected error occurred", _request_id(request) + ), + ) diff --git a/src/api/middleware.py b/src/api/middleware.py new file mode 100644 index 0000000..7fd87d9 --- /dev/null +++ b/src/api/middleware.py @@ -0,0 +1,38 @@ +"""Per-request correlation id (ADR-0008, ADR-0011). + +Every request gets a `request_id`: reused from an incoming `X-Request-Id` if +the caller supplied one, otherwise generated. It is bound into structlog's +contextvars so every log line emitted while handling the request carries it, +stored on `request.state` for exception handlers, and echoed back in the +response header. +""" + +import uuid +from collections.abc import Awaitable, Callable +from typing import override + +import structlog +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request +from starlette.responses import Response + +_HEADER = "X-Request-Id" + + +class RequestIdMiddleware(BaseHTTPMiddleware): + @override + async def dispatch( + self, request: Request, call_next: Callable[[Request], Awaitable[Response]] + ) -> Response: + request_id = request.headers.get(_HEADER) or str(uuid.uuid4()) + request.state.request_id = request_id + + structlog.contextvars.clear_contextvars() + structlog.contextvars.bind_contextvars(request_id=request_id) + try: + response = await call_next(request) + finally: + structlog.contextvars.clear_contextvars() + + response.headers[_HEADER] = request_id + return response diff --git a/src/api/router.py b/src/api/router.py index af9233c..ea50999 100644 --- a/src/api/router.py +++ b/src/api/router.py @@ -1,3 +1,6 @@ from fastapi import APIRouter +from src.api.routers.files import router as files_router + router = APIRouter() +router.include_router(files_router) diff --git a/src/api/routers/files.py b/src/api/routers/files.py new file mode 100644 index 0000000..8d5e7f5 --- /dev/null +++ b/src/api/routers/files.py @@ -0,0 +1,74 @@ +"""`POST /v1/files`, `GET /v1/files/{file_id}` (ADR-0008). + +Routes adapt HTTP to `application/files` calls; they do not parse, hash, +touch MinIO/Qdrant, or otherwise carry ingestion business logic (ADR-0015). +""" + +import uuid +from typing import Annotated + +from anyio import CapacityLimiter +from fastapi import APIRouter, Depends, Form, HTTPException, Response, UploadFile, status +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.api.dependencies.auth import require_scope +from src.api.schemas.files import FileStatusResponse, FileUploadResponse +from src.application.auth.context import AuthContext +from src.application.files.status import get_file_status +from src.application.files.upload import upload_source_file +from src.application.ports.object_storage import ObjectStorage +from src.bootstrap.dependencies import ( + get_ingestion_limiter, + get_object_storage, + get_sessionmaker, + get_settings, +) +from src.config import Settings + +router = APIRouter(prefix="/files", tags=["files"]) + +_RequireFilesWrite = Annotated[AuthContext, Depends(require_scope("files:write"))] +_SessionmakerDep = Annotated[async_sessionmaker[AsyncSession], Depends(get_sessionmaker)] +_ObjectStorageDep = Annotated[ObjectStorage, Depends(get_object_storage)] +_SettingsDep = Annotated[Settings, Depends(get_settings)] +_IngestionLimiterDep = Annotated[CapacityLimiter, Depends(get_ingestion_limiter)] + + +@router.post("", status_code=status.HTTP_201_CREATED) +async def upload_file( + response: Response, + file: UploadFile, + domain: Annotated[str, Form()], + auth: _RequireFilesWrite, + sessionmaker: _SessionmakerDep, + storage: _ObjectStorageDep, + settings: _SettingsDep, + limiter: _IngestionLimiterDep, +) -> FileUploadResponse: + data = await file.read() + result = await upload_source_file( + sessionmaker=sessionmaker, + storage=storage, + auth=auth, + domain=domain, + filename=file.filename or "", + data=data, + max_upload_size_bytes=settings.ingestion.max_upload_size_bytes, + chunking_strategy=settings.chunking.strategy, + validation_limiter=limiter, + ) + if not result.is_new_attempt: + response.status_code = status.HTTP_200_OK + return FileUploadResponse.from_result(result) + + +@router.get("/{file_id}") +async def get_file( + file_id: uuid.UUID, + auth: _RequireFilesWrite, + sessionmaker: _SessionmakerDep, +) -> FileStatusResponse: + result = await get_file_status(sessionmaker, tenant_id=auth.tenant_id, source_file_id=file_id) + if result is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="file not found") + return FileStatusResponse.from_result(result) diff --git a/src/api/schemas/errors.py b/src/api/schemas/errors.py new file mode 100644 index 0000000..c28a962 --- /dev/null +++ b/src/api/schemas/errors.py @@ -0,0 +1,16 @@ +"""The ADR-0008 error envelope.""" + +from typing import Any + +from pydantic import BaseModel + + +class ErrorDetail(BaseModel): + code: str + message: str + details: dict[str, Any] = {} + request_id: str | None = None + + +class ErrorResponse(BaseModel): + error: ErrorDetail diff --git a/src/api/schemas/files.py b/src/api/schemas/files.py new file mode 100644 index 0000000..a62c921 --- /dev/null +++ b/src/api/schemas/files.py @@ -0,0 +1,50 @@ +"""Public request/response models for `/v1/files` (ADR-0008). + +Separate from the SQLAlchemy ORM models and the `application/files` domain +dataclasses (ADR-0015): this is the shape callers see. +""" + +import uuid + +from pydantic import BaseModel + +from src.application.files.models import UploadResult +from src.application.files.status import FileStatusResult + + +class FileUploadResponse(BaseModel): + file_id: uuid.UUID + ingestion_job_id: uuid.UUID + status: str + chunks_indexed: int + + @classmethod + def from_result(cls, result: UploadResult) -> "FileUploadResponse": + return cls( + file_id=result.file_id, + ingestion_job_id=result.ingestion_job_id, + status=result.status, + chunks_indexed=result.chunks_indexed, + ) + + +class FileStatusResponse(BaseModel): + file_id: uuid.UUID + source_filename: str + domain: str + status: str + ingestion_job_id: uuid.UUID | None + ingestion_status: str | None + chunks_indexed: int + + @classmethod + def from_result(cls, result: FileStatusResult) -> "FileStatusResponse": + return cls( + file_id=result.file_id, + source_filename=result.source_filename, + domain=result.domain, + status=result.status, + ingestion_job_id=result.ingestion_job_id, + ingestion_status=result.ingestion_status, + chunks_indexed=result.chunks_indexed, + ) diff --git a/src/bootstrap/dependencies.py b/src/bootstrap/dependencies.py index 4f0f82c..a1aaabe 100644 --- a/src/bootstrap/dependencies.py +++ b/src/bootstrap/dependencies.py @@ -1,11 +1,13 @@ from collections.abc import AsyncIterator from dataclasses import dataclass +from anyio import CapacityLimiter from fastapi import Request from minio import Minio from qdrant_client import AsyncQdrantClient from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker +from src.application.ports.object_storage import ObjectStorage from src.config import Settings @@ -16,6 +18,8 @@ class AppResources: db_sessionmaker: async_sessionmaker[AsyncSession] minio_client: Minio qdrant_client: AsyncQdrantClient + object_storage: ObjectStorage + ingestion_limiter: CapacityLimiter def _resources(request: Request) -> AppResources: @@ -34,6 +38,25 @@ def get_qdrant_client(request: Request) -> AsyncQdrantClient: return _resources(request).qdrant_client +def get_object_storage(request: Request) -> ObjectStorage: + return _resources(request).object_storage + + +def get_ingestion_limiter(request: Request) -> CapacityLimiter: + return _resources(request).ingestion_limiter + + +def get_sessionmaker(request: Request) -> async_sessionmaker[AsyncSession]: + """The session *factory*, not a request-scoped session. + + Application services that own more than one transaction in a single + request (ADR-0017's two-phase upload) need to open and close sessions + themselves rather than borrow one request-scoped session that would + otherwise stay open across the whole request. + """ + return _resources(request).db_sessionmaker + + async def get_db_session(request: Request) -> AsyncIterator[AsyncSession]: sessionmaker = _resources(request).db_sessionmaker async with sessionmaker() as session: diff --git a/src/bootstrap/lifespan.py b/src/bootstrap/lifespan.py index 03edcf7..73a69e9 100644 --- a/src/bootstrap/lifespan.py +++ b/src/bootstrap/lifespan.py @@ -2,13 +2,14 @@ from collections.abc import AsyncIterator, Callable from contextlib import AbstractAsyncContextManager, asynccontextmanager import structlog -from anyio import to_thread +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 @@ -43,12 +44,22 @@ def create_lifespan( 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: diff --git a/src/main.py b/src/main.py index 93d3381..0fd8aca 100644 --- a/src/main.py +++ b/src/main.py @@ -1,5 +1,7 @@ from fastapi import FastAPI +from src.api.errors import register_exception_handlers +from src.api.middleware import RequestIdMiddleware from src.api.router import router as v1_router from src.api.routers.health import router as health_router from src.bootstrap.lifespan import create_lifespan @@ -8,6 +10,8 @@ from src.config import Settings def create_app(settings: Settings | None = None) -> FastAPI: app = FastAPI(lifespan=create_lifespan(settings)) + app.add_middleware(RequestIdMiddleware) + register_exception_handlers(app) app.include_router(health_router) app.include_router(v1_router, prefix="/v1") return app