Blueprint: HTTP API service¶

from pydantic import BaseModel
from pydantic_settings import BaseSettings, SettingsConfigDict


class LoggingSettings(BaseModel):
    level: str = "INFO"
    use_json: bool = True


class MetricsSettings(BaseModel):
    enabled: bool = False        # the API exposes /system/metrics on its own port
    host: str = "0.0.0.0"
    port: int = 9090
    prefix: str | None = None


class HttpSettings(BaseModel):
    host: str = "0.0.0.0"
    port: int = 8000
    cors_origins: list[str] = []


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_nested_delimiter="__", env_file=".env")

    app_version: str = "0.0.0"
    environment: str = "local"
    database_dsn: str

    http: HttpSettings = HttpSettings()
    logging: LoggingSettings = LoggingSettings()
    metrics: MetricsSettings = MetricsSettings()
    tracing: None = None
    error_tracking: None = None

    def get_app_version(self) -> str:
        return self.app_version

from collections.abc import AsyncIterator

from dishka import Provider, Scope, provide
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine


class DatabaseProvider(Provider):
    def __init__(self, dsn: str) -> None:
        super().__init__()
        self._dsn = dsn

    @provide(scope=Scope.APP)
    async def engine(self) -> AsyncIterator[AsyncEngine]:
        engine = create_async_engine(self._dsn, pool_size=10, pool_pre_ping=True)
        yield engine
        await engine.dispose()            # runs when the app scope closes — last

    @provide(scope=Scope.APP)
    def session_maker(self, engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
        return async_sessionmaker(engine, expire_on_commit=False)

    @provide(scope=Scope.REQUEST)
    async def session(self, maker: async_sessionmaker[AsyncSession]) -> AsyncIterator[AsyncSession]:
        async with maker() as session:
            yield session                 # closed when the unit scope closes


class UseCaseProvider(Provider):
    scope = Scope.REQUEST

    @provide
    def orders(self, session: AsyncSession) -> OrderRepository:
        return PostgresOrderRepository(session)

    @provide
    def get_order(self, orders: OrderRepository) -> GetOrderUseCase:
        return GetOrderUseCase(orders)

from dishka import make_async_container

from servicewright.adapters.dishka import DishkaContainer


def build_container(settings: Settings) -> DishkaContainer:
    return DishkaContainer(
        make_async_container(
            DatabaseProvider(settings.database_dsn),
            UseCaseProvider(),
        )
    )

from servicewright import ErrorKind, ServiceError


class OrderNotFoundError(ServiceError):
    kind = ErrorKind.NOT_FOUND          # code → "order_not_found", HTTP 404


class OrderAlreadyPaidError(ServiceError):
    kind = ErrorKind.CONFLICT           # code → "order_already_paid", HTTP 409


class PaymentGatewayError(ServiceError):
    kind = ErrorKind.UNAVAILABLE
    public = False                      # masked: the client never sees the reason

from fastapi import APIRouter

from servicewright.adapters.fastapi import UnitScopeDep

router = APIRouter(prefix="/orders", tags=["orders"])


@router.get("/{order_id}", response_model=OrderResponse)
async def get_order(order_id: UUID, scope: UnitScopeDep) -> OrderResponse:
    use_case = await scope.get(GetOrderUseCase)
    order = await use_case.execute(order_id)
    return OrderResponse.model_validate(order)


@router.post("", response_model=OrderResponse, status_code=201)
async def create_order(payload: CreateOrderRequest, scope: UnitScopeDep) -> OrderResponse:
    use_case = await scope.get(CreateOrderUseCase)
    order = await use_case.execute(payload.to_dto())
    return OrderResponse.model_validate(order)

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from servicewright import AppSpec, AppScopeProtocol, KeyRedactor, ObsConfig, ObservabilityManager, ServiceContext
from servicewright.adapters.health.postgres import PostgresHealthCheck
from servicewright.adapters.warmers.postgres import PostgresWarmer


def build_spec(settings: Settings) -> AppSpec:
    spec = AppSpec(
        service_name="orders-service",
        create_container=build_container,
        observability=ObservabilityManager(
            ObsConfig(metrics="prometheus", logging="structlog"),
            redactor=KeyRedactor(),
        ),
        drain_grace_seconds=30.0,
        cleanup_timeout_seconds=10.0,
    )

    async def build_warmers(ctx: ServiceContext) -> list:
        """Runs with the app scope open, before readiness."""
        maker = await ctx.app_scope.get(async_sessionmaker[AsyncSession])
        return [PostgresWarmer(SessionManagerAdapter(maker), timeout=10.0)]

    async def register_health_checks(app_scope: AppScopeProtocol | None = None) -> None:
        """pre_start runs after warmup, before bind — checks exist before the first probe."""
        maker = await app_scope.get(async_sessionmaker[AsyncSession])
        spec.health.add_check("postgres", PostgresHealthCheck(maker, timeout=5.0))

    spec.warmers_factory = build_warmers
    spec.lifecycle.add_pre_start_hook(register_health_checks)
    return spec

from servicewright.adapters.fastapi import CORSMiddlewareConfig, FastApiEntrypoint, HttpConfig, MiddlewareConfig


def build_http(settings: Settings) -> FastApiEntrypoint:
    return FastApiEntrypoint(
        config=HttpConfig(
            host=settings.http.host,
            port=settings.http.port,
            version=settings.app_version,
            graceful_timeout=10.0,
        ),
        routers=(orders_router,),
        middlewares=MiddlewareConfig(
            cors=CORSMiddlewareConfig(allow_origins=settings.http.cors_origins),
        ),
        metrics=True,
    )

import asyncio

from servicewright import Service


def main() -> None:
    settings = Settings()
    service = Service(build_spec(settings), entrypoints=[build_http(settings)])
    asyncio.run(service.run(settings))


if __name__ == "__main__":
    main()