CRUD over Postgres

A DocumentSpec plus a PostgresDepsModule is the entire persistence story. The FastAPI routes call a DocumentFacade and return read models — no persistence ports, no SQL, no ORM, and optimistic concurrency for free.

The runnable version lives at examples/recipes/crud_fastapi/ — just run brings up ephemeral Postgres, serves the API, and tears it down.

The aggregate¶

A Product with the three write shapes — domain model, create command, and a partial update — plus a read model:

class Product(Document):
    name: str
    price: int


class ProductCreate(CreateDocumentCmd):
    name: str
    price: int


class ProductUpdate(BaseDTO):
    name: str | None = None
    price: int | None = None


class ProductRead(ReadDocument):
    name: str
    price: int

The specification¶

One spec names the aggregate and its write types. "products" is the logical name shared by the adapter wiring:

product_spec = DocumentSpec(
    name="products",
    read=ProductRead,
    write=DocumentWriteTypes(
        domain=Product, create_cmd=ProductCreate, update_cmd=ProductUpdate
    ),
)

Wire Postgres¶

PostgresDocumentConfig maps the spec to its tables; PostgresDepsModule registers the document ports under "products", and the lifecycle module owns the connection pool:

def build_runtime(pg: PostgresClient, *, dsn: str) -> ExecutionRuntime:
    deps = DepsRegistry.from_modules(
        PostgresDepsModule(
            client=pg, rw_documents={"products": PRODUCT_PG}, tx={"products"}
        ),
    )
    lifecycle = LifecyclePlan.from_modules(
        PostgresLifecycleModule(client=pg, dsn=dsn, config=PostgresConfig()),
    )
    return ExecutionRuntime(deps=deps.freeze(), lifecycle=lifecycle.freeze())

The demo table

The example creates its products table on startup so it's self-contained. A real service owns its schema through migrations — Forze reads and writes rows, it doesn't manage DDL. The columns id, rev, created_at, and last_update_at are the document bookkeeping fields.

The routes¶

The registry turns the spec and boundary DTOs into typed operations. The runtime opens inside the app's lifespan; each route calls the resulting facade:

# DTOs are derived from the spec's read + write models.
registry = build_document_registry(product_spec).freeze()


@asynccontextmanager
async def lifespan(app: FastAPI):
    pg = PostgresClient()
    dsn = os.environ.get("POSTGRES_DSN", "postgresql://forze:forze@localhost:5432/forze")
    _rt.set_once(build_runtime(pg, dsn=dsn))
    async with _rt.get().scope():
        await pg.execute(SCHEMA)  # demo bootstrap (real apps migrate instead)
        yield


app = FastAPI(title="Products API", lifespan=lifespan)
register_exception_handlers(app)  # CoreException → HTTP (not_found → 404, conflict → 409)


def products() -> DocumentFacade[ProductRead, ProductCreate, ProductUpdate]:
    return DocumentFacade(
        ctx=ctx(),
        registry=registry,
        namespace=product_spec.default_namespace,
    )


@app.post("/products")
async def create_product(cmd: ProductCreate) -> ProductRead:
    return await products().create(cmd)


@app.get("/products/{product_id}")
async def get_product(product_id: UUID) -> ProductRead:
    return await products().get(DocumentIdDTO(id=product_id))


@app.get("/products")
async def list_products() -> list[ProductRead]:
    page = await products().list(ListRequestDTO())
    return list(page.hits)


@app.put("/products/{product_id}")
async def update_product(product_id: UUID, rev: int, patch: ProductUpdate) -> ProductRead:
    result = await products().update(
        DocumentUpdateDTO(id=product_id, rev=rev, dto=patch)
    )
    return result.data


@app.delete("/products/{product_id}", status_code=204)
async def delete_product(product_id: UUID) -> None:
    await products().kill(DocumentIdDTO(id=product_id))

Create / get / list / delete map straight onto facade operations.
Update carries the document's rev — a stale rev raises a conflict, which register_exception_handlers turns into a 409. That's optimistic concurrency with no extra code.
A missing id raises not_found → 404.

Hand-writing the routes keeps this recipe transparent; the same endpoints can also be generated from an operation registry with attach_document_routes.

Run it¶

cd examples/recipes/crud_fastapi
just run

Then open http://localhost:8000/docs.