Domain

The central registry object. Create one per bounded context to register all domain elements, manage configuration, and coordinate infrastructure adapters.

See Compose a Domain for a practical guide.

The domain object is a one-stop gateway to:

Registering Domain Objects/Concepts
Querying/Retrieving Domain Artifacts like Entities, Services, etc.
Retrieve injected infrastructure adapters

Usually you create a Domain instance in your main module or in the __init__.py file of your package like this::

from protean import Domain
domain = Domain()

The Domain will automatically detect the root path of the calling module. You can also specify the root path explicitly::

domain = Domain(root_path="/path/to/domain")

The root path resolution follows this priority:

Explicit root_path parameter if provided
DOMAIN_ROOT_PATH environment variable if set
Auto-detection of caller's file location
Current working directory as last resort

:param root_path: the path to the folder containing the domain file (optional, will auto-detect if not provided) :param name: the name of the domain (optional, will use the module name if not provided) :param config: optional configuration dictionary :param identity_function: optional function to generate identities for domain objects

Source code in src/protean/domain/__init__.py

def __init__(
    self,
    root_path: str | None = None,
    name: str = "",
    config: dict[str, Any] | None = None,
    identity_function: Callable[..., Any] | None = None,
) -> None:
    self.root_path: str
    # Determine root_path based on resolution priority
    if root_path is None:
        # Try to get from environment variable
        env_root_path = os.environ.get("DOMAIN_ROOT_PATH")
        if env_root_path:
            self.root_path = env_root_path
        else:
            # Auto-detect
            self.root_path = self._guess_caller_path()
    else:
        self.root_path = root_path

    # Initialize the domain with the name of the module if not provided
    # Get the stack frame of the caller of the __init__ method
    caller_frame = inspect.stack()[1]
    # Get the module name from the globals of the frame where the object was instantiated
    self.name = name if name else caller_frame.frame.f_globals["__name__"]

    # Registry for all domain Objects
    self._domain_registry = _DomainRegistry()

    #: The configuration dictionary as ``Config``.  This behaves
    #: exactly like a regular dictionary but supports additional methods
    #: to load a config from files.
    self.config: Config2 = self.load_config(config)

    # The function to invoke to generate identity
    self._identity_function = identity_function

    # Injectable source of "now" for deadline, lock, and retry-backoff time
    # comparisons. Defaults to real UTC; tests assign a stub clock to make
    # boundary behavior deterministic. See :class:`protean.utils.Clock`.
    self.clock: Clock = SystemClock()

    self.providers: Providers = Providers(self)
    self.event_store: EventStore = EventStore(self)
    self.brokers: Brokers = Brokers(self)
    self.caches: Caches = Caches(self)
    self.email_providers: EmailProviders = EmailProviders(self)

    # Cache for holding Model to Entity/Aggregate associations
    # Structure mirrors Providers._repositories:
    # {
    #    'app.User': {
    #        'postgresql': UserPostgresModel,
    #        'sqlite': UserSQLiteModel,
    #        None: UserGenericModel,       # database=None → fallback for any provider
    #    }
    # }
    self._database_models: dict[str, dict[str | None, type[BaseDatabaseModel]]] = (
        defaultdict(dict)
    )
    self._constructed_models: dict[str, BaseDatabaseModel] = {}

    # Message enricher hooks — callables that add custom metadata to events/commands.
    # Event enrichers receive (event, aggregate) and return dict[str, Any].
    # Command enrichers receive (command,) and return dict[str, Any].
    # Results are merged into metadata.extensions.
    self._event_enrichers: list[Callable[..., Any]] = []
    self._command_enrichers: list[Callable[..., Any]] = []
    # Aggregate pre-persist enrichers receive (aggregate,) and mutate it
    # in place to stamp cross-cutting lifecycle/audit fields on save.
    self._aggregate_enrichers: list[Callable[..., Any]] = []

    # Composed helpers — see handler_setup.py, validation.py, etc.
    self._command_processor = CommandProcessor(self)
    self._handler_configurator = HandlerConfigurator(self)
    self._infrastructure = InfrastructureManager(self)
    self._query_processor = QueryProcessor(self)
    self._resolver = ElementResolver(self)
    self._type_manager = TypeManager(self)
    self._validator = DomainValidator(self)

    #: A list of functions that are called when the domain context
    #: is destroyed.  This is the place to store code that cleans up and
    #: disconnects from databases, for example.
    self.teardown_domain_context_functions: list[Callable[..., Any]] = []

    # Placeholder array for resolving classes referenced by domain elements
    self._pending_class_resolutions: dict[str, Any] = defaultdict(list)

    # Partition-per-key routing map (ADR-0028): stream_category ->
    # partition-key field name. Built by ``HandlerConfigurator`` during
    # ``_prepare()`` from every handler that declares ``sequential_by``, and
    # read by the Unit of Work at commit to denormalize the key onto each
    # outbox row. Empty when no handler opts in.
    self._partition_keys: dict[str, str] = {}

    # Event classes that have already emitted a raise-time deprecation
    # warning, so a deprecated event warns once per type, not per instance.
    self._deprecated_events_warned: set[type] = set()

    # Lazy-initialized idempotency store
    self._idempotency_store: IdempotencyStore | None = None

    # Lazy-initialized trace emitter for command processing observability
    self._trace_emitter: TraceEmitter | None = None

    # Lazy-initialized OpenTelemetry providers (set by init_telemetry)
    self._otel_tracer_provider: Any = None
    self._otel_meter_provider: Any = None
    self._otel_init_attempted = False

has_outbox `property`

has_outbox: bool

Whether the outbox pattern is active.

Derived from server.default_subscription_type: outbox is enabled when subscription type is "stream". For backward compatibility, an explicit enable_outbox = true also activates the outbox.

camel_case_name `cached` `property`

camel_case_name: str

Return the CamelCase name of the domain.

The CamelCase name is the name of the domain with the first letter capitalized.

Examples:

my_domain -> MyDomain
my_domain_1 -> MyDomain1
my_domain_1_0 -> MyDomain10

normalized_name `cached` `property`

normalized_name: str

Return the normalized name of the domain.

The normalized name is the underscored version of the domain name.

Examples:

MyDomain -> my_domain
My Domain -> my_domain
My-Domain -> my_domain
My Domain 1 -> my_domain_1
My Domain 1.0 -> my_domain_1_0

init

init(traverse: bool = True) -> None

Parse the domain folder, and attach elements dynamically to the domain.

Protean parses all files in the domain file's folder, as well as under it, to load elements. So, all domain files are to be nested under the file contain the domain definition.

One can use the traverse flag to control this functionality, True by default.

When enabled, Protean is responsible for loading domain elements and ensuring all functionality is activated.

The developer is responsible for activating functionality manually when autoloading is disabled. Element activation can be done by importing them in central areas of domain execution, like Application Services.

For example, asynchronous aspects of a domain like its Subscribers and Event Handlers should be imported in their relevant Application Services and Aggregates.

This method bubbles up circular import issues, if present, in the domain code.

Source code in src/protean/domain/__init__.py

def init(self, traverse: bool = True) -> None:
    """Parse the domain folder, and attach elements dynamically to the domain.

    Protean parses all files in the domain file's folder, as well as under it,
    to load elements. So, all domain files are to be nested under the file contain
    the domain definition.

    One can use the `traverse` flag to control this functionality, `True` by default.

    When enabled, Protean is responsible for loading domain elements and ensuring
    all functionality is activated.

    The developer is responsible for activating functionality manually when
    autoloading is disabled. Element activation can be done by importing them
    in central areas of domain execution, like Application Services.

    For example, asynchronous aspects of a domain like its Subscribers and
    Event Handlers should be imported in their relevant Application Services
    and Aggregates.

    This method bubbles up circular import issues, if present, in the domain code.
    """
    self._auto_configure_logging()

    self._prepare(traverse=traverse)

    # Initialize adapters after loading domain
    self._initialize()

    # Gate sequential_by handlers on broker STREAM_PARTITIONING support
    # (ADR-0028); needs live brokers, so it runs after _initialize().
    self._handler_configurator.validate_sequential_by_capabilities()

    # Initialize outbox DAOs for all providers
    if self.has_outbox:
        self._initialize_outbox()

    # Initialize consume-side idempotency markers when any projector opts in
    if self.has_idempotent_consumers:
        self._initialize_processed_messages()

domain_context

domain_context(**kwargs: Any) -> DomainContext

Create a DomainContext. Use as a with block to push the context, which will make current_domain point at this domain.

::

with domain.domain_context():
    init_db()

Source code in src/protean/domain/__init__.py

def domain_context(self, **kwargs: Any) -> DomainContext:
    """Create a ``DomainContext``. Use as a ``with``
    block to push the context, which will make ``current_domain``
    point at this domain.

    ::

        with domain.domain_context():
            init_db()
    """
    return DomainContext(self, **kwargs)

aggregate

aggregate(_cls: type[_T]) -> type[_T]

aggregate(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

aggregate(
    _cls: type[_T" backlink-type="used-by" backlink-anchor="protean.domain.Domain.aggregate" optional hover>_T] | None = None, **kwargs: Any
) -> type[_T" backlink-type="returned-by" backlink-anchor="protean.domain.Domain.aggregate" optional hover>_T] | Callable[[type[_T" backlink-type="returned-by" backlink-anchor="protean.domain.Domain.aggregate" optional hover>_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def aggregate(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.AGGREGATE,
        _cls=_cls,
        **kwargs,
    )

entity

entity(_cls: type[_T]) -> type[_T]

entity(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

entity(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def entity(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(DomainObjects.ENTITY, _cls=_cls, **kwargs)

value_object

value_object(_cls: type[_T]) -> type[_T]

value_object(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

value_object(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def value_object(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.VALUE_OBJECT,
        _cls=_cls,
        **kwargs,
    )

command

command(_cls: type[_T]) -> type[_T]

command(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

command(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def command(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.COMMAND,
        _cls=_cls,
        **kwargs,
    )

event

event(_cls: type[_T]) -> type[_T]

event(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

event(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def event(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.EVENT,
        _cls=_cls,
        **kwargs,
    )

command_handler

command_handler(_cls: type[_T]) -> type[_T]

command_handler(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

command_handler(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def command_handler(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(DomainObjects.COMMAND_HANDLER, _cls=_cls, **kwargs)

event_handler

event_handler(_cls: type[_T]) -> type[_T]

event_handler(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

event_handler(
    _cls: type[_T" backlink-type="used-by" backlink-anchor="protean.domain.Domain.event_handler" optional hover>_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def event_handler(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.EVENT_HANDLER,
        _cls=_cls,
        **kwargs,
    )

application_service

application_service(_cls: type[_T]) -> type[_T]

application_service(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

application_service(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def application_service(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.APPLICATION_SERVICE,
        _cls=_cls,
        **kwargs,
    )

domain_service

domain_service(_cls: type[_T]) -> type[_T]

domain_service(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

domain_service(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def domain_service(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.DOMAIN_SERVICE,
        _cls=_cls,
        **kwargs,
    )

repository

repository(_cls: type[_T]) -> type[_T]

repository(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

repository(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def repository(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(DomainObjects.REPOSITORY, _cls=_cls, **kwargs)

projection

projection(_cls: type[_T]) -> type[_T]

projection(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

projection(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def projection(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.PROJECTION,
        _cls=_cls,
        **kwargs,
    )

projector

projector(_cls: type[_T]) -> type[_T]

projector(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

projector(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def projector(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.PROJECTOR,
        _cls=_cls,
        **kwargs,
    )

subscriber

subscriber(_cls: type[_T]) -> type[_T]

subscriber(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

subscriber(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def subscriber(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.SUBSCRIBER,
        _cls=_cls,
        **kwargs,
    )

process_manager

process_manager(_cls: type[_T]) -> type[_T]

process_manager(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

process_manager(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def process_manager(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(
        DomainObjects.PROCESS_MANAGER,
        _cls=_cls,
        **kwargs,
    )

upcaster

upcaster(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Register an event upcaster with the domain.

Upcasters transform raw event payloads from an old schema version to a newer one. They are applied lazily during deserialization so that @apply handlers and event handlers always see the current schema.

PARAMETER	DESCRIPTION
`event_type`	The event class this upcaster targets (current version). TYPE: `type`
`from_version`	Source version number (e.g. `1`). TYPE: `int`
`to_version`	Target version number (e.g. `2`). TYPE: `int`

Example::

@domain.upcaster(event_type=OrderPlaced, from_version=1, to_version=2)
class UpcastOrderPlacedV1ToV2(BaseUpcaster):
    def upcast(self, data: dict) -> dict:
        data["currency"] = "USD"
        return data

Source code in src/protean/domain/__init__.py

def upcaster(
    self,
    _cls: type[_T] | None = None,
    **kwargs: Any,
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    """Register an event upcaster with the domain.

    Upcasters transform raw event payloads from an old schema version to
    a newer one.  They are applied lazily during deserialization so that
    ``@apply`` handlers and event handlers always see the current schema.

    Keyword Args:
        event_type (type): The event class this upcaster targets (current version).
        from_version (int): Source version number (e.g. ``1``).
        to_version (int): Target version number (e.g. ``2``).

    Example::

        @domain.upcaster(event_type=OrderPlaced, from_version=1, to_version=2)
        class UpcastOrderPlacedV1ToV2(BaseUpcaster):
            def upcast(self, data: dict) -> dict:
                data["currency"] = "USD"
                return data
    """
    return self._domain_element(
        DomainObjects.UPCASTER,
        _cls=_cls,
        **kwargs,
    )

database_model

database_model(_cls: type[_T]) -> type[_T]

database_model(
    _cls: None = ..., **kwargs: Any
) -> Callable[[type[_T]], type[_T]]

database_model(
    _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]

Source code in src/protean/domain/__init__.py

@dataclass_transform(
    field_specifiers=(
        Auto,
        Boolean,
        Date,
        DateTime,
        DictField,
        Float,
        HasMany,
        HasOne,
        Identifier,
        Integer,
        ListField,
        Reference,
        Status,
        String,
        Text,
        ValueObject,
    )
)
def database_model(
    self, _cls: type[_T] | None = None, **kwargs: Any
) -> type[_T] | Callable[[type[_T]], type[_T]]:
    return self._domain_element(DomainObjects.DATABASE_MODEL, _cls=_cls, **kwargs)

process

process(
    command: Any,
    asynchronous: bool | None = None,
    idempotency_key: str | None = None,
    raise_on_duplicate: bool = False,
    priority: int | None = None,
    correlation_id: str | None = None,
    deadline: datetime | None = None,
    timeout: timedelta | None = None,
) -> Any | None

Process command and return results based on specified preference.

By default, Protean does not return values after processing commands. This behavior can be overridden either by setting command_processing in config to "sync" or by specifying asynchronous=False when calling the domain's handle method.

PARAMETER	DESCRIPTION
`command`	Command to process (instance of a `@domain.command`-decorated class) TYPE: `Any`
`asynchronous`	Specifies if the command should be processed asynchronously. Defaults to True. TYPE: `Boolean` DEFAULT: `None`
`idempotency_key`	Caller-provided key for command deduplication. When provided, enables submission-level dedup via the idempotency store. TYPE: `str` DEFAULT: `None`
`raise_on_duplicate`	If `True`, raises `DuplicateCommandError` when a duplicate idempotency key is detected. If `False` (default), silently returns the cached result. TYPE: `bool` DEFAULT: `False`
`priority`	Processing priority for events produced by this command. TYPE: `int` DEFAULT: `None`
`correlation_id`	Correlation ID for distributed tracing. TYPE: `str` DEFAULT: `None`
`deadline`	Absolute time after which the command must not be executed. Stored in metadata and propagated to downstream commands. Mutually exclusive with `timeout`. TYPE: `datetime` DEFAULT: `None`
`timeout`	Relative deadline (`now + timeout`), converted to an absolute deadline at submission. Mutually exclusive with `deadline`. TYPE: `timedelta` DEFAULT: `None`

RETURNS	DESCRIPTION
`Any \| None`	Optional[Any]: Returns either the command handler's return value or nothing, based on preference.

Source code in src/protean/domain/__init__.py

def process(
    self,
    command: Any,
    asynchronous: bool | None = None,
    idempotency_key: str | None = None,
    raise_on_duplicate: bool = False,
    priority: int | None = None,
    correlation_id: str | None = None,
    deadline: datetime | None = None,
    timeout: timedelta | None = None,
) -> Any | None:
    """Process command and return results based on specified preference.

    By default, Protean does not return values after processing commands. This behavior
    can be overridden either by setting command_processing in config to "sync" or by specifying
    ``asynchronous=False`` when calling the domain's ``handle`` method.

    Args:
        command: Command to process (instance of a ``@domain.command``-decorated class)
        asynchronous (Boolean, optional): Specifies if the command should be processed asynchronously.
            Defaults to True.
        idempotency_key (str, optional): Caller-provided key for command deduplication.
            When provided, enables submission-level dedup via the idempotency store.
        raise_on_duplicate (bool): If ``True``, raises ``DuplicateCommandError``
            when a duplicate idempotency key is detected. If ``False`` (default),
            silently returns the cached result.
        priority (int, optional): Processing priority for events produced by this command.
        correlation_id (str, optional): Correlation ID for distributed tracing.
        deadline (datetime, optional): Absolute time after which the command must not
            be executed. Stored in metadata and propagated to downstream commands.
            Mutually exclusive with ``timeout``.
        timeout (timedelta, optional): Relative deadline (``now + timeout``), converted
            to an absolute deadline at submission. Mutually exclusive with ``deadline``.

    Returns:
        Optional[Any]: Returns either the command handler's return value or nothing, based on preference.
    """
    return self._command_processor.process(
        command,
        asynchronous=asynchronous,
        idempotency_key=idempotency_key,
        raise_on_duplicate=raise_on_duplicate,
        priority=priority,
        correlation_id=correlation_id,
        deadline=deadline,
        timeout=timeout,
    )

Domain

has_outbox property

camel_case_name cached property

normalized_name cached property

init

domain_context

aggregate

entity

value_object

command

event

command_handler

event_handler

application_service

domain_service

repository

projection

projector

subscriber

process_manager

upcaster

database_model

process

has_outbox `property`

camel_case_name `cached` `property`

normalized_name `cached` `property`