feat(bootstrap,api,infra): scaffold app composition, health route, and postgres/minio/qdrant adapters

Why:
- first runnable slice of ADR-0012's resource-lifetime rules and ADR-0015's package layout: app-lifetime clients built once in the lifespan, released via explicit dependencies.

Changes:
- Settings split into per-domain nested settings (postgres/minio/ingestion/qdrant/app/logging); FastAPI app wired with /healthz, /readyz and a /v1 router; Postgres/MinIO/Qdrant adapters and SQLAlchemy models for tenants, API keys, source files, ingestion jobs/events.
This commit is contained in:
2026-08-16 11:54:14 +03:30
parent df221279a5
commit 3c660de093
28 changed files with 616 additions and 2 deletions

0
src/__init__.py Normal file
View File

0
src/api/__init__.py Normal file
View File

View File

3
src/api/router.py Normal file
View File

@@ -0,0 +1,3 @@
from fastapi import APIRouter
router = APIRouter()

View File

36
src/api/routers/health.py Normal file
View File

@@ -0,0 +1,36 @@
import asyncio
from fastapi import APIRouter, Request, Response, status
from src.bootstrap.dependencies import AppResources
from src.infrastructure.minio.client import ping as ping_minio
from src.infrastructure.postgres.database import ping as ping_postgres
from src.infrastructure.qdrant.client import ping as ping_qdrant
router = APIRouter(tags=["health"])
@router.get("/healthz")
async def healthz() -> dict[str, str]:
return {"status": "ok"}
@router.get("/readyz")
async def readyz(request: Request, response: Response) -> dict[str, bool]:
resources: AppResources = request.app.state.resources
timeout = resources.settings.app.readiness_check_timeout_seconds
postgres_ready, minio_ready, qdrant_ready = await asyncio.gather(
ping_postgres(resources.db_engine, timeout),
ping_minio(resources.minio_client, timeout),
ping_qdrant(resources.qdrant_client, timeout),
)
result = {
"postgres": postgres_ready,
"minio": minio_ready,
"qdrant": qdrant_ready,
}
if not all(result.values()):
response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
return result

View File

View File

View File

@@ -0,0 +1,44 @@
from collections.abc import AsyncIterator
from dataclasses import dataclass
from fastapi import Request
from minio import Minio
from qdrant_client import AsyncQdrantClient
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
from src.config import Settings
@dataclass
class AppResources:
settings: Settings
db_engine: AsyncEngine
db_sessionmaker: async_sessionmaker[AsyncSession]
minio_client: Minio
qdrant_client: AsyncQdrantClient
def _resources(request: Request) -> AppResources:
return request.app.state.resources
def get_settings(request: Request) -> Settings:
return _resources(request).settings
def get_minio_client(request: Request) -> Minio:
return _resources(request).minio_client
def get_qdrant_client(request: Request) -> AsyncQdrantClient:
return _resources(request).qdrant_client
async def get_db_session(request: Request) -> AsyncIterator[AsyncSession]:
sessionmaker = _resources(request).db_sessionmaker
async with sessionmaker() as session:
try:
yield session
except Exception:
await session.rollback()
raise

56
src/bootstrap/lifespan.py Normal file
View File

@@ -0,0 +1,56 @@
from collections.abc import AsyncIterator, Callable
from contextlib import AbstractAsyncContextManager, asynccontextmanager
import structlog
from fastapi import FastAPI
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.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)
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")
app.state.resources = AppResources(
settings=resolved_settings,
db_engine=db_engine,
db_sessionmaker=db_sessionmaker,
minio_client=minio_client,
qdrant_client=qdrant_client,
)
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

View File

@@ -1,5 +1,76 @@
from pydantic_settings import BaseSettings from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
class PostgresSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="POSTGRES_", extra="ignore")
host: str = "127.0.0.1"
port: int = 5433
user: str = "chatbot"
password: str = "chatbot"
db: str = "chatbot"
@property
def dsn(self) -> str:
return f"postgresql+asyncpg://{self.user}:{self.password}@{self.host}:{self.port}/{self.db}"
class MinioSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="MINIO_", extra="ignore")
endpoint: str = "127.0.0.1:9100"
access_key: str = "chatbot"
secret_key: str = "chatbot-secret"
secure: bool = False
bucket: str = "chatbot-source-files"
class IngestionSettings(BaseSettings):
"""Bounds on inline ingestion (ADR-0017).
`timeout_seconds` must stay below the proxy/load-balancer/client read
timeouts, or callers give up on work that is still succeeding.
"""
model_config = SettingsConfigDict(env_prefix="INGESTION_", extra="ignore")
max_concurrency: int = 4
thread_pool_size: int = 8
timeout_seconds: float = 120.0
max_chunks_per_file: int = 5000
embed_batch_size: int = 128
embed_concurrency: int = 4
class QdrantSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="QDRANT_", extra="ignore")
url: str = "http://127.0.0.1:6343"
api_key: str | None = None
class AppLimitSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="APP_", extra="ignore")
env: str = "local"
max_upload_size_mb: int = 25
readiness_check_timeout_seconds: float = 2.0
class LoggingSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="LOG_", extra="ignore")
level: str = "INFO"
json_format: bool = False
class Settings(BaseSettings): class Settings(BaseSettings):
pass model_config = SettingsConfigDict(env_file=".env", extra="ignore")
postgres: PostgresSettings = Field(default_factory=PostgresSettings)
minio: MinioSettings = Field(default_factory=MinioSettings)
ingestion: IngestionSettings = Field(default_factory=IngestionSettings)
qdrant: QdrantSettings = Field(default_factory=QdrantSettings)
app: AppLimitSettings = Field(default_factory=AppLimitSettings)
logging: LoggingSettings = Field(default_factory=LoggingSettings)

View File

View File

View File

@@ -0,0 +1,23 @@
import asyncio
from minio import Minio
from src.config import MinioSettings
def create_client(settings: MinioSettings) -> Minio:
return Minio(
settings.endpoint,
access_key=settings.access_key,
secret_key=settings.secret_key,
secure=settings.secure,
)
async def ping(client: Minio, timeout: float) -> bool:
try:
async with asyncio.timeout(timeout):
await asyncio.to_thread(client.bucket_exists, "healthcheck-probe")
except Exception:
return False
return True

View File

@@ -0,0 +1,84 @@
import logging
import logging.config
import sys
import structlog
from src.config import LoggingSettings
def configure_logging(settings: LoggingSettings) -> None:
shared_processors = [
structlog.contextvars.merge_contextvars,
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.processors.TimeStamper(fmt="iso", utc=True),
structlog.processors.StackInfoRenderer(),
]
structlog.configure(
processors=[
*shared_processors,
structlog.processors.format_exc_info,
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
renderer = (
structlog.processors.JSONRenderer()
if settings.json_format
else structlog.dev.ConsoleRenderer(colors=True)
)
logging.config.dictConfig(
{
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"default": {
"()": structlog.stdlib.ProcessorFormatter,
"processors": [
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
renderer,
],
"foreign_pre_chain": [
structlog.stdlib.ExtraAdder(),
*shared_processors,
],
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": settings.level,
"formatter": "default",
"stream": sys.stdout,
},
},
"loggers": {
"": {
"handlers": ["console"],
"level": settings.level,
"propagate": False,
},
"uvicorn": {
"handlers": ["console"],
"level": settings.level,
"propagate": False,
},
"uvicorn.access": {
"handlers": ["console"],
"level": settings.level,
"propagate": False,
},
"sqlalchemy.engine": {
"handlers": ["console"],
"level": "WARNING",
"propagate": False,
},
},
}
)

View File

View File

@@ -0,0 +1,28 @@
import asyncio
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
AsyncEngine,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from src.config import PostgresSettings
def create_engine(settings: PostgresSettings) -> AsyncEngine:
return create_async_engine(settings.dsn)
def create_sessionmaker(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
return async_sessionmaker(engine, expire_on_commit=False)
async def ping(engine: AsyncEngine, timeout: float) -> bool:
try:
async with asyncio.timeout(timeout), engine.connect() as connection:
await connection.execute(text("SELECT 1"))
except Exception:
return False
return True

View File

@@ -0,0 +1,15 @@
from src.infrastructure.postgres.models.api_key import ApiKey
from src.infrastructure.postgres.models.base import Base
from src.infrastructure.postgres.models.ingestion_job import IngestionJob
from src.infrastructure.postgres.models.ingestion_job_event import IngestionJobEvent
from src.infrastructure.postgres.models.source_file import SourceFile
from src.infrastructure.postgres.models.tenant import Tenant
__all__ = [
"ApiKey",
"Base",
"IngestionJob",
"IngestionJobEvent",
"SourceFile",
"Tenant",
]

View File

@@ -0,0 +1,41 @@
import uuid
from datetime import datetime
from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Index, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from src.infrastructure.postgres.models.base import Base
API_KEY_ACTOR_TYPES = ("backend", "admin", "worker")
API_KEY_STATUSES = ("active", "revoked", "expired")
class ApiKey(Base):
__tablename__ = "api_keys"
__table_args__ = (
CheckConstraint(f"actor_type IN {API_KEY_ACTOR_TYPES}", name="ck_api_keys_actor_type"),
CheckConstraint(f"status IN {API_KEY_STATUSES}", name="ck_api_keys_status"),
Index("ix_api_keys_tenant_id_status", "tenant_id", "status"),
)
id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
tenant_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("tenants.id", ondelete="CASCADE"), index=True
)
name: Mapped[str] = mapped_column(String(200))
key_prefix: Mapped[str] = mapped_column(String(32), unique=True, index=True)
key_hash: Mapped[str] = mapped_column(String(255))
scopes: Mapped[list[str]] = mapped_column(JSONB, default=list, server_default="[]")
actor_type: Mapped[str] = mapped_column(String(20), default="backend", server_default="backend")
status: Mapped[str] = mapped_column(String(20), default="active", server_default="active")
expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
created_by: Mapped[str | None] = mapped_column(String(200), default=None)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)

View File

@@ -0,0 +1,5 @@
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass

View File

@@ -0,0 +1,59 @@
import uuid
from datetime import datetime
from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Index, Integer, String, Text, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from src.infrastructure.postgres.models.base import Base
INGESTION_JOB_STATUSES = ("queued", "running", "succeeded", "failed", "cancelled")
INGESTION_CHUNKING_STRATEGIES = ("semantic", "fixed_size")
class IngestionJob(Base):
__tablename__ = "ingestion_jobs"
__table_args__ = (
CheckConstraint(f"status IN {INGESTION_JOB_STATUSES}", name="ck_ingestion_jobs_status"),
CheckConstraint(
f"chunking_strategy IN {INGESTION_CHUNKING_STRATEGIES}",
name="ck_ingestion_jobs_chunking_strategy",
),
Index("ix_ingestion_jobs_tenant_id_status_created_at", "tenant_id", "status", "created_at"),
Index("ix_ingestion_jobs_source_file_id_created_at", "source_file_id", "created_at"),
)
id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
tenant_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("tenants.id", ondelete="CASCADE"), index=True
)
source_file_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("source_files.id", ondelete="CASCADE"), index=True
)
requested_by_api_key_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("api_keys.id", ondelete="SET NULL"), default=None
)
status: Mapped[str] = mapped_column(String(20), default="queued", server_default="queued")
chunking_strategy: Mapped[str | None] = mapped_column(String(20), default=None)
embedding_model_versions: Mapped[dict[str, object]] = mapped_column(
JSONB, default=dict, server_default="{}"
)
started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)
points_created: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
points_updated: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
points_soft_deleted: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
points_skipped: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
error_code: Mapped[str | None] = mapped_column(String(100), default=None)
error_message: Mapped[str | None] = mapped_column(Text, default=None)
metadata_: Mapped[dict[str, object]] = mapped_column(
"metadata", JSONB, default=dict, server_default="{}"
)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)

View File

@@ -0,0 +1,47 @@
import uuid
from datetime import datetime
from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Index, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from src.infrastructure.postgres.models.base import Base
INGESTION_JOB_EVENT_LEVELS = ("info", "warning", "error")
INGESTION_JOB_EVENT_STAGES = (
"received",
"parsed",
"chunked",
"embedded",
"upserted",
"completed",
)
class IngestionJobEvent(Base):
__tablename__ = "ingestion_job_events"
__table_args__ = (
CheckConstraint(
f"level IN {INGESTION_JOB_EVENT_LEVELS}", name="ck_ingestion_job_events_level"
),
CheckConstraint(
f"stage IN {INGESTION_JOB_EVENT_STAGES}", name="ck_ingestion_job_events_stage"
),
Index(
"ix_ingestion_job_events_ingestion_job_id_created_at", "ingestion_job_id", "created_at"
),
)
id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
tenant_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("tenants.id", ondelete="CASCADE"), index=True
)
ingestion_job_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("ingestion_jobs.id", ondelete="CASCADE"), index=True
)
level: Mapped[str] = mapped_column(String(10))
stage: Mapped[str] = mapped_column(String(20))
message: Mapped[str] = mapped_column(String(1000))
details: Mapped[dict[str, object]] = mapped_column(JSONB, default=dict, server_default="{}")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

View File

@@ -0,0 +1,41 @@
import uuid
from datetime import datetime
from sqlalchemy import BigInteger, CheckConstraint, DateTime, ForeignKey, Index, String, func
from sqlalchemy.orm import Mapped, mapped_column
from src.infrastructure.postgres.models.base import Base
SOURCE_FILE_TYPES = ("csv", "xlsx", "docx", "doc")
SOURCE_FILE_STATUSES = ("active", "superseded", "soft_deleted", "purged")
class SourceFile(Base):
__tablename__ = "source_files"
__table_args__ = (
CheckConstraint(f"source_type IN {SOURCE_FILE_TYPES}", name="ck_source_files_source_type"),
CheckConstraint(f"status IN {SOURCE_FILE_STATUSES}", name="ck_source_files_status"),
Index("ix_source_files_tenant_id_domain_created_at", "tenant_id", "domain", "created_at"),
Index("ix_source_files_tenant_id_content_sha256", "tenant_id", "content_sha256"),
)
id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
tenant_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("tenants.id", ondelete="CASCADE"), index=True
)
domain: Mapped[str] = mapped_column(String(80))
source_filename: Mapped[str] = mapped_column(String(500))
source_type: Mapped[str] = mapped_column(String(10))
content_sha256: Mapped[str] = mapped_column(String(64))
byte_size: Mapped[int] = mapped_column(BigInteger)
storage_uri: Mapped[str | None] = mapped_column(String(1000), default=None)
status: Mapped[str] = mapped_column(String(20), default="active", server_default="active")
created_by_api_key_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("api_keys.id", ondelete="SET NULL"), default=None
)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)

View File

@@ -0,0 +1,27 @@
import uuid
from datetime import datetime
from sqlalchemy import CheckConstraint, DateTime, String, func
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from src.infrastructure.postgres.models.base import Base
TENANT_STATUSES = ("active", "suspended", "deleted")
class Tenant(Base):
__tablename__ = "tenants"
__table_args__ = (CheckConstraint(f"status IN {TENANT_STATUSES}", name="ck_tenants_status"),)
id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
slug: Mapped[str] = mapped_column(String(80), unique=True, index=True)
name: Mapped[str] = mapped_column(String(200))
status: Mapped[str] = mapped_column(String(20), default="active", server_default="active")
settings: Mapped[dict[str, object]] = mapped_column(JSONB, default=dict, server_default="{}")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
)
deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None)

View File

View File

@@ -0,0 +1,18 @@
import asyncio
from qdrant_client import AsyncQdrantClient
from src.config import QdrantSettings
def create_client(settings: QdrantSettings) -> AsyncQdrantClient:
return AsyncQdrantClient(url=settings.url, api_key=settings.api_key)
async def ping(client: AsyncQdrantClient, timeout: float) -> bool:
try:
async with asyncio.timeout(timeout):
await client.get_collections()
except Exception:
return False
return True

View File

@@ -0,0 +1,16 @@
from fastapi import FastAPI
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
from src.config import Settings
def create_app(settings: Settings | None = None) -> FastAPI:
app = FastAPI(lifespan=create_lifespan(settings))
app.include_router(health_router)
app.include_router(v1_router, prefix="/v1")
return app
app = create_app()