Skip to main content

Domain Module Overview

info

The Domain module is the core of Fluvius Framework, providing CQRS, event sourcing, and domain-driven design capabilities.

Introduction​

The Domain module implements Domain-Driven Design patterns including:

  • Aggregates: Business logic containers
  • Commands: State change requests
  • Events: Immutable facts
  • State Management: Current state queries
  • Event Sourcing: Event log storage

Key Components​

Domain​

The Domain class is the main entry point for domain logic:

from fluvius.domain import Domain

class UserDomain(Domain):
    __aggregate__ = UserAggregate
    __namespace__ = 'app-user'

Aggregate​

Aggregates contain business logic and manage state:

from fluvius.domain import Aggregate, action

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

Command​

Commands represent intentions to change state:

command = domain.create_command('create-user', {
    'name': 'John Doe',
    'email': 'john@example.com'
})

Event​

Events are generated automatically from aggregate actions:

@action(evt_key='user-created')
async def create_user(self, name: str, email: str):
    # Event 'user-created' is generated automatically
    pass

State Manager​

The State Manager provides read access to current state:

# Fetch entity
user = await domain.statemgr.fetch('user', user_id)

# Query entities
users = await domain.statemgr.find('user', active=True)

Quick Example​

import asyncio
from fluvius.domain import Domain, Aggregate, action
from fluvius.domain.context import SanicContext
from fluvius.data import UUID_GENR, DataModel, field

# Define state
class UserState(DataModel):
    name: str = field()
    email: str = field()
    active: bool = field(initial=True)

# Define aggregate
class UserAggregate(Aggregate):
    def __init__(self, domain):
        super().__init__(domain)
        self._state = None
    
    @action(evt_key='user-created', resources=['user'])
    async def create_user(self, name: str, email: str):
        self._state = UserState(name=name, email=email, active=True)
        return self._state

# Define domain
class UserDomain(Domain):
    __aggregate__ = UserAggregate

# Use domain
async def main():
    ctx = SanicContext.create(namespace='app-user')
    domain = UserDomain(ctx)
    
    user_id = UUID_GENR()
    domain.set_aggroot('user', user_id)
    
    command = domain.create_command('create-user', {
        'name': 'John Doe',
        'email': 'john@example.com'
    })
    
    response = await domain.process_command(command)
    print(response)

asyncio.run(main())

Features​

  • CQRS: Separate command and query paths
  • Event Sourcing: Automatic event generation and storage
  • State Management: Built-in state manager for queries
  • Authorization: Integrated with Casbin for policy enforcement
  • Transactions: ACID-compliant transaction support
  • Logging: Comprehensive activity and event logging

Next Steps​