Forze provides a declarative system for composing usecases with middleware. Instead of manually wiring guards and effects into every operation, you declare them in plans and registries that are resolved at runtime.

How composition works

The composition model has three parts:

1. Registry: maps operation names to usecase factories
2. Plan: describes which middleware (guards, effects, transactions) wraps each operation
3. Facade: ties registry, plan, and spec together into a single entry point

When a request arrives, the facade resolves the usecase factory from the registry, wraps it with the middleware chain from the plan, and returns a callable ready to execute.

Operation registry

The UsecaseRegistry maps operation keys to usecase factories. Each factory receives an ExecutionContext and returns a Usecase instance:

from forze.application.execution import UsecaseRegistry


registry = UsecaseRegistry()
registry = registry.register("get", lambda ctx: GetProject(ctx=ctx))
registry = registry.register("create", lambda ctx: CreateProject(ctx=ctx))

For document aggregates, build_document_registry(spec, dtos) creates a registry pre-populated with standard CRUD operations (GET, CREATE, UPDATE, KILL, DELETE, RESTORE).

Usecase plan

The UsecasePlan describes how each operation is composed with middleware. It maps operation keys to middleware buckets that run at specific stages.

Plan buckets

Each operation has seven middleware buckets, executed in this order:

The in_tx_* and after_commit buckets only activate when tx=True for the operation.

Building a plan

from forze.application.execution import UsecasePlan


plan = (
    UsecasePlan()
    .tx("create")
    .tx("update")
    .before("create", auth_guard, priority=100)
    .after("create", log_effect, priority=0)
    .after_commit("create", notify_effect)
    .before("*", rate_limit_guard, priority=200)
)

The wildcard "*" applies to all operations as a base plan. Per-operation plans extend the base. When resolved, the base and operation-specific plans are merged.

Priority ordering

Middlewares within a bucket are sorted by priority (descending). Higher priority runs first (outermost). Priority values must be unique within a bucket to avoid ambiguity.

Merging plans

Multiple plans can be merged for modular composition:

base_plan = tx_document_plan
auth_plan = build_auth_plan()
audit_plan = build_audit_plan()

final_plan = UsecasePlan.merge(base_plan, auth_plan, audit_plan)

Inspecting a plan

Use explain() to see the resolved middleware chain for an operation:

explanation = plan.explain("create")
print(explanation.pretty_format())

This outputs the full chain with bucket names, priorities, and factory references, useful for debugging composition issues.

Document composition

For document aggregates, Forze provides a pre-built composition layer:

from forze.application.composition.document import (
    DocumentDTOs,
    DocumentUsecasesFacade,
    build_document_registry,
    tx_document_plan,
)

project_dtos = DocumentDTOs(
    read=ProjectReadModel,
    create=CreateProjectCmd,
    update=UpdateProjectCmd,
)

registry = build_document_registry(project_spec, project_dtos)
registry.extend_plan(tx_document_plan, inplace=True)

build_document_registry(spec, dtos) registers standard usecase factories for all DocumentOperation variants.

tx_document_plan is a pre-built UsecasePlan with transaction wrapping for all operations.

Create a facade from an execution context and the registry:

facade = DocumentUsecasesFacade(ctx=ctx, reg=registry)
project = await facade.create(CreateProjectCmd(title="New"))
fetched = await facade.get(project.id)

Document operations

The DocumentOperation enum defines the standard operation keys:

Extending document composition

Add custom middleware to the default plan:

from forze.application.composition.document import (
    DocumentOperation,
    tx_document_plan,
)


def my_auth_guard(ctx):
    async def guard(args):
        if not is_authorized(ctx):
            raise PermissionError("Not authorized")
    return guard


plan = (
    tx_document_plan
    .before(DocumentOperation.CREATE, my_auth_guard, priority=100)
    .before(DocumentOperation.UPDATE, my_auth_guard, priority=100)
    .after_commit(DocumentOperation.CREATE, my_notification_effect)
)

Search composition

Search follows the same pattern:

from forze.application.composition.search import (
    SearchDTOs,
    SearchUsecasesFacade,
    build_search_registry,
)

search_dtos = SearchDTOs(read=ProjectReadModel)
search_registry = build_search_registry(project_search_spec, search_dtos)

facade = SearchUsecasesFacade(ctx=ctx, reg=search_registry)
result = await facade.search(
    SearchRequestDTO(query="roadmap", limit=20)
)

DTO mapping

The mapping pipeline transforms incoming DTOs before they reach the usecase:

from forze.application.mapping import DTOMapper, NumberIdStep, CreatorIdStep

mapper = (
    build_document_create_mapper(project_spec, project_dtos)
    .with_steps(NumberIdStep(), CreatorIdStep())
)

Each MappingStep can inject computed fields (like number_id from a counter or creator_id from the actor context) into the DTO before it reaches the create usecase.

Custom usecases

You can register entirely custom usecases alongside the standard ones:

from forze.application.execution import Usecase, UsecaseRegistry


class ArchiveProject(Usecase[UUID, ProjectReadModel]):
    async def main(self, args: UUID) -> ProjectReadModel:
        doc = self.ctx.doc_write(project_spec)
        return await doc.update(args, UpdateProjectCmd(status="archived"))


registry = build_document_registry(project_spec, project_dtos)
registry = registry.register("archive", lambda ctx: ArchiveProject(ctx=ctx))

plan = (
    tx_document_plan
    .tx("archive")
    .before("archive", auth_guard, priority=100)
)

The custom operation integrates into the same composition system as built-in operations and benefits from the same middleware infrastructure.