Quickstart

What you will build¶

A minimal REST service for a User aggregate:

Storage is in-memory — no Docker, no migrations. The complete, runnable file is examples/quickstart/app.py; the steps below build it up.

Step 1 — Create the project¶

uv init forze-quickstart
cd forze-quickstart
uv add 'forze[fastapi]'

Everything below goes into a single main.py.

Step 2 — Define the domain models¶

An aggregate needs a domain model, a create command, and a read model. Document gives you id, rev, and timestamps for free.

class User(Document):
    name: str
    email: str | None = None


class CreateUserCmd(CreateDocumentCmd):
    name: str
    email: str | None = None


class ReadUser(ReadDocument):
    name: str
    email: str | None = None

    @computed_field
    @property
    def email_provided(self) -> bool:
        return self.email is not None

Step 3 — Declare a specification¶

The specification is the logical name — "users" — that ties the models to their operations and, later, to adapters.

user_spec = DocumentSpec(
    name="users",
    read=ReadUser,
    write={
        "domain": User,
        "create_cmd": CreateUserCmd,
    },
)

Step 4 — Build the operation registry¶

build_document_registry assembles the standard CRUD operations; freeze() makes the registry immutable and shareable.

registry = build_document_registry(
    user_spec, DocumentDTOs(read=ReadUser, create=CreateUserCmd)
).freeze()

Step 5 — Wire the runtime¶

MockDepsModule provides in-memory adapters for every contract. build_runtime assembles an ExecutionRuntime around it — the runtime builds the context on startup, and runtime.get_context() reaches it per request.

runtime = build_runtime(MockDepsModule())

Step 6 — Attach the routes¶

runtime_lifespan runs the runtime inside the app's lifespan. Each route resolves a DocumentFacade from the context and calls an operation — the handlers never touch HTTP:

app = FastAPI(title="Users API", lifespan=runtime_lifespan(runtime))
register_exception_handlers(app)  # CoreException → HTTP (e.g. not_found → 404)


@app.post("/users")
async def create_user(cmd: CreateUserCmd) -> ReadUser:
    return await users().create(cmd)


@app.get("/users/{user_id}")
async def get_user(user_id: UUID) -> ReadUser:
    return await users().get(DocumentIdDTO(id=user_id))


@app.get("/users")
async def list_users() -> list[ReadUser]:
    page = await runtime.get_context().document.query(user_spec).find_many()
    return list(page.hits)


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

register_exception_handlers maps a CoreException to a response, so a missing user comes back as a 404. (Routes are hand-wired here to show the moving parts — generated routes can attach the CRUD endpoints from the registry instead.)

Step 7 — Run it¶

uv run uvicorn main:app --reload

Open http://localhost:8000/docs for the interactive explorer, or try it from the shell:

# Create — note the id in the response
curl -s -X POST http://127.0.0.1:8000/users \
  -H 'Content-Type: application/json' \
  -d '{"name": "Ada", "email": "ada@example.com"}'

curl -s http://127.0.0.1:8000/users            # list
curl -s http://127.0.0.1:8000/users/<id>       # get one
curl -s -X DELETE http://127.0.0.1:8000/users/<id>   # delete

What you just did¶

You built a complete service without a single line of HTTP or storage code in your domain:

• A User aggregate with its command and read models — pure Python, no infrastructure.
• A specification and a frozen operation registry — the named operations the service exposes.
• An ExecutionRuntime wired to in-memory adapters, opened for the app's lifetime.
• Routes that resolve operations from the context and return read models.

The only thing tying this to "in-memory" is MockDepsModule in Step 5. Swap it for PostgresDepsModule + RedisDepsModule and the domain, spec, registry, and routes don't change — that's the whole point. The PostgreSQL integration shows the swap.

Where to go next¶