csrd-service 0.5.6

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

Consumer Feature Matrix (BaseService + ServiceError)

Capability area	Common expectation	Actual support	How to use / notes
Request context access	Service can read current user and request id	✅ Built-in	BaseService.current_user() and BaseService.current_request_id()
Service base class	Lightweight base for domain services	✅ Built-in	Inherit from BaseService; inject repo/delegate dependencies in your constructor
Domain error taxonomy	Business errors map to consistent HTTP statuses	✅ Built-in	Use NotFoundError (404), ConflictError (409), ValidationError (422), AuthorizationError (403), DownstreamError (502)
Structured API error responses	Service exceptions become API error payloads	✅ Built-in (with registration)	Register service_exception_handler for ServiceError
Per-error status override	Override default status code per raise	✅ Built-in	raise ValidationError(..., status_code=409)
Error metadata	Include detail/code fields for clients	✅ Built-in	detail and code are preserved by the handler
Logging integration	Auto logging from service methods	⚙️ Via composition	Mix in csrd.logging.LoggingMixin; not part of BaseService itself
Transport-agnostic domain logic	Avoid HTTPException in business layer	✅ Built-in pattern	Raise ServiceError subclasses and keep HTTP concerns in handler layer
Dependency injection wiring	FastAPI DI helper exists in package	⚙️ Pattern-based	Use your own factory + Depends; package documents the pattern
Transactions / unit-of-work	Automatic transactional boundaries	❌ Not built-in	Handle in repository/DB layer
Idempotency utilities	Built-in idempotency keys/replay protection	❌ Not built-in	Implement per endpoint/service policy
Policy engine / RBAC framework	Full authorization policy runtime	❌ Not built-in	Use claim checks in service code and/or external policy layer

Dependency tier

Tier 1 (standalone)     csrd.models · csrd.lifespan · csrd.context
Tier 2 (uses Tier 1)    csrd.delegate  csrd.repository  csrd.service
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