Skip to main content

Domain Module API Reference

info

Complete API reference for the Domain module classes and functions.

Domain​

The main domain class that orchestrates command processing and event handling.

Class: Domain​

class Domain(DomainSignalManager, DomainEntityRegistry):
    __namespace__ = None
    __aggregate__ = None
    __statemgr__ = StateManager
    __logstore__ = DomainLogStore
    __revision__ = 0
    __config__ = SimpleNamespace
    __context__ = DomainContextData
    __policymgr__ = None

Methods​

create_command(cmd_key, payload, aggroot=None)​

Create a command bundle.

Parameters:

  • cmd_key (str): Command identifier
  • payload (dict | DataModel): Command payload
  • aggroot (tuple): Aggregate root (resource, identifier, domain_sid, domain_iid)

Returns: CommandBundle

process_command(command)​

Process a single command.

Parameters:

  • command (CommandBundle): Command to process

Returns: DomainResponse

set_aggroot(resource, identifier, domain_sid=None, domain_iid=None)​

Set the aggregate root for command processing.

Parameters:

  • resource (str): Resource name
  • identifier (UUID_TYPE): Resource identifier
  • domain_sid (UUID_TYPE, optional): Domain scope ID
  • domain_iid (UUID_TYPE, optional): Domain item ID

Aggregate​

Base class for domain aggregates containing business logic.

Class: Aggregate​

class Aggregate:
    def __init__(self, domain):
        self.domain_name = domain.domain_name
        self.lookup_event = domain.lookup_event
        self.lookup_message = domain.lookup_message
        self.lookup_response = domain.lookup_response
        self.statemgr = domain.statemgr

Decorator: @action(evt_key, resources)​

Decorator for aggregate methods that generate events.

Parameters:

  • evt_key (str): Event identifier
  • resources (str | list): Allowed resources

Example:

@action(evt_key='user-created', resources=['user'])
async def create_user(self, name: str, email: str):
    # Business logic
    pass

Command​

Command bundle containing command data.

Class: CommandBundle​

class CommandBundle(DomainEntityRecord):
    domain: str
    command: str
    revision: int
    resource: str
    identifier: UUID_TYPE
    payload: DataModel | BlankModel
    domain_sid: UUID_TYPE | None
    domain_iid: UUID_TYPE | None
    context: UUID_TYPE | None
    status: CommandState

Event​

Event record generated from aggregate actions.

Class: EventRecord​

class EventRecord(DomainEntityRecord):
    event: str
    src_cmd: UUID_TYPE
    args: dict
    data: dict | BlankModel | DataModel

State Manager​

Provides read access to domain state.

Class: StateManager​

class StateManager(DataAccessManager):
    pass

Methods​

fetch(resource, identifier)​

Fetch a single entity by ID.

Parameters:

  • resource (str): Resource name
  • identifier (UUID_TYPE): Entity identifier

Returns: DataModel | None

find(resource, **filters)​

Find entities matching filters.

Parameters:

  • resource (str): Resource name
  • **filters: Filter criteria

Returns: list[DataModel]

find_one(resource, **filters)​

Find a single entity matching filters.

Parameters:

  • resource (str): Resource name
  • **filters: Filter criteria

Returns: DataModel | None

query(query)​

Execute a query.

Parameters:

  • query (BackendQuery): Query object

Returns: list[DataModel]

Context​

Domain context providing request and environment information.

Class: DomainContext​

class DomainContext:
    request: dict
    transport: DomainTransport
    source: str
    realm: str
    user_id: UUID_TYPE | None
    organization_id: UUID_TYPE | None
    profile_id: UUID_TYPE | None

Class: SanicContext​

Context factory for Sanic/FastAPI applications.

Method: create(namespace, **kwargs)​

Create a domain context.

Parameters:

  • namespace (str): Domain namespace
  • **kwargs: Additional context data

Returns: DomainContext

Next Steps​