csrd-service 0.1.43

Service layer base class with domain error hierarchy

csrd.service

Service-layer boilerplate for the CSRD pattern (Controller → Service → Repository / Delegate).

What it provides

Component	Purpose
BaseService	Lightweight base class with per-class logger and request-context accessors (current_user(), current_request_id())
ServiceError	Base domain exception — keeps business logic free of HTTPException
NotFoundError	404 — resource doesn't exist
ConflictError	409 — duplicate, version mismatch, etc.
ValidationError	422 — business-rule validation (not schema validation)
AuthorizationError	403 — authenticated user lacks permission
DownstreamError	502 — a delegate call to another service failed
service_exception_handler	FastAPI handler that maps any ServiceError → structured APIErrorResponse JSON

Dependency tier

Tier 1 (standalone)     csrd.models · csrd.lifespan · csrd.context
Tier 2 (uses Tier 1)    csrd.delegate · csrd.repository · csrd.service  ← NEW
Tier 3 (uses Tier 1+2)  csrd.versioning

csrd.service depends on csrd.context and csrd.models only. It does not import csrd.repository or csrd.delegate — your services accept those as constructor params.

Quick start

from csrd.service import BaseService, NotFoundError, DownstreamError


class OrderService(BaseService):
    def __init__(self, repo: OrderRepository, payments: PaymentsDelegate):
        super().__init__()
        self._repo = repo
        self._payments = payments

    async def get_order(self, order_id: int) -> Order:
        order = await self._repo.get_by_id(order_id)
        if order is None:
            raise NotFoundError("Order not found", detail=f"order_id={order_id}")
        return order

    async def place_order(self, cart: Cart) -> Order:
        order = await self._repo.create(cart)
        try:
            await self._payments.charge(order)
        except Exception as exc:
            raise DownstreamError("Payment failed", cause=exc) from exc
        return order

Register the exception handler

With csrd.versioning:

from csrd.service import ServiceError, service_exception_handler
from csrd.versioning import VersionedApiConfig

configure_versioned_api(
    app,
    version_mapping=VERSIONS,
    config=VersionedApiConfig(
        ex_handlers=[(ServiceError, service_exception_handler)],
    ),
)

Or on a plain FastAPI app:

app.add_exception_handler(ServiceError, service_exception_handler)

Wire into FastAPI DI

from functools import cache
from typing import Annotated
from fastapi import Depends

@cache
def order_service_factory(repo: OrderRepository, payments: PaymentsDelegate):
    return OrderService(repo, payments)

OrderServiceDep = Annotated[OrderService, Depends(order_service_factory)]

# In your endpoint:
@router.get("/orders/{order_id}")
async def get_order(service: OrderServiceDep, order_id: int):
    return await service.get_order(order_id)

Installation

uv add csrd-service  # or: pip install csrd-service