When to compose

DDD CQRS ES

The Domain class in Protean acts as a composition root. It manages external dependencies and injects them into objects during application startup.

Your domain should be composed at the start of the application lifecycle, once, before any request or task is handled. This means:

1. Instantiate the Domain and register elements (via decorators or manual registration).
2. Initialize the domain with domain.init() to activate adapters, validate element registration, and resolve dependencies.
3. Push a domain context before processing requests, so that current_domain is available throughout the call stack.

The exact integration point depends on your application framework.

FastAPI (recommended)

Protean provides built-in middleware for FastAPI that pushes and pops the domain context for each request automatically. This example defines the domain in the same file. In a larger app you would import it from its own module instead:

from fastapi import FastAPI

from protean import Domain
from protean.fields import String
from protean.integrations.fastapi import (
    DomainContextMiddleware,
    register_exception_handlers,
)

domain = Domain(name="Tasks")


@domain.aggregate
class Task:
    title: String(max_length=200, required=True)


# Initialize the domain at module load time
domain.init(traverse=False)

app = FastAPI()

# Middleware pushes/pops domain context per request
app.add_middleware(
    DomainContextMiddleware,
    route_domain_map={"/": domain},
)

# Map domain exceptions to HTTP status codes
register_exception_handlers(app)

The middleware runs every request inside a domain context. The exception handlers translate ValidationError, ObjectNotFoundError, etc. into appropriate HTTP responses.

See FastAPI Integration for the full guide.

Flask

For Flask, use before_request and teardown_request hooks to manage the domain context. Keep the context you push on Flask's g, so the teardown hook pops that same context:

from flask import Flask, g

from protean import Domain
from protean.domain.context import has_domain_context
from protean.fields import Integer, String

domain = Domain()


@domain.aggregate
class User:
    first_name: String(max_length=50)
    last_name: String(max_length=50)
    age: Integer()


def create_app(config):
    app = Flask(__name__, static_folder=None)

    domain.config.from_object(config)

    domain.init(traverse=False)

    @app.before_request
    def push_context():
        if not has_domain_context():
            # Push up a Domain Context and keep it to pop later
            g.domain_context = domain.domain_context()
            g.domain_context.push()

    @app.teardown_request
    def pop_context(exc):
        # Pop the Domain Context this request pushed, even on an error
        context = g.pop("domain_context", None)
        if context is not None:
            context.pop(exc)

    return app

The domain is initialized once during create_app(). The context is pushed before each request and popped when the request ends, even if it fails.

Console applications and scripts

In simple console applications, compose the domain in a main function and use a with block for the domain context:

from protean import Domain
from protean.fields import String

domain = Domain()


@domain.aggregate
class Task:
    title: String(max_length=200, required=True)


def main():
    domain.init(traverse=False)

    with domain.domain_context():
        repo = domain.repository_for(Task)
        repo.add(Task(title="Write documentation"))


if __name__ == "__main__":
    main()

Background workers and the Protean server

The Protean server (protean server) composes the domain internally. You only need to point it at your domain module:

$ protean server --domain my_app.domain

The server initializes the domain, pushes a context, and runs the event processing loop. See Running the Server for details.

Key principles

• Compose once, early: Call domain.init() at startup, not per-request. Initialization activates database connections, validates element registration, and is not designed to be called repeatedly.
• Context per request: Each request or task should run inside its own domain context (domain.domain_context()). The context makes current_domain available and manages the Unit of Work lifecycle.
• Let the framework manage it: Prefer framework-provided integration (FastAPI middleware, Flask hooks) over manual context management. The framework integration cleans up each context, even when an exception occurs.