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 handles domain context management automatically:

from fastapi import FastAPI
from protean.integrations.fastapi import (
    DomainContextMiddleware,
    register_exception_handlers,
)

from my_app.domain import domain

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

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 ensures every request runs inside a domain context, and 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 after_request hooks to manage the domain context:

import logging.config

from flask import Flask

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)
    logging.config.dictConfig(domain.config["LOGGING_CONFIG"])

    domain.init()

    @app.before_request
    def set_context():
        if not has_domain_context():
            # Push up a Domain Context
            domain.domain_context().push()

    @app.after_request
    def pop_context(response):
        # Pop the Domain Context
        domain.domain_context().pop()

        return response

    return app

The domain is initialized once during create_app(), and the context is pushed before each request and popped after.

Console applications and scripts

In simple console applications, compose the domain in main 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)

if __name__ == "__main__":
    domain.init()

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

Background workers and the Protean server

The Protean server (protean server) handles domain composition 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 manages 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. This ensures contexts are properly cleaned up even when exceptions occur.