Domain Module

fluvius.domain provides the write side of Fluvius: domains register command, event, response, and message types; aggregates implement mutations; state managers persist resource state; log stores retain command activity.

Define an aggregate

from fluvius.data import serialize_mapping
from fluvius.domain import Aggregate
from fluvius.domain.aggregate import action


class UserAggregate(Aggregate):
    @action("user-created", resources="user")
    async def create_user(self, data):
        record = self.init_resource(
            "user",
            **serialize_mapping(data),
            _id=self.aggroot.identifier,
        )
        await self.statemgr.insert(record)
        return record

Bind a domain

from fluvius.domain import Domain, MessageDispatcher, SQLDomainLogStore


class UserDomain(Domain):
    __namespace__ = "user-profile"
    __aggregate__ = UserAggregate
    __statemgr__ = UserStateManager
    __logstore__ = SQLDomainLogStore
    __msgdispatcher__ = MessageDispatcher

Register commands

from fluvius.data import DataModel


class CreateUser(UserDomain.Command):
    class Meta:
        key = "create-user"
        name = "Create User"
        resource_init = True
        auth_required = True
        policy_required = True
        roles_required = ("editor",)

    class Data(DataModel):
        name: str

    async def _process(self, aggregate, statemgr, payload):
        user = await aggregate.create_user(payload)
        yield aggregate.create_response({"_id": str(user._id)})

Command metadata controls authentication, policy, roles, allowed methods/zones, resource initialization, and MCP exposure. Role checks use profile roles; IAM realm roles are reserved for API-zone authorization.

Commands are dispatched through mounted HTTP endpoints or work-queue clients. The old pattern of instantiating a domain with SanicContext and calling create_command is not part of the current API.