Skip to main content

New Module

info

Learn how to create a new domain module in Fluvius Framework.

Creating a New Domain​

Step 1: Create Domain Folder​

Create a folder for your domain:

mkdir -p src/myapp/domains/user
cd src/myapp/domains/user

Step 2: Define State Model​

Create state.py:

from fluvius.data import DataModel, field
from datetime import datetime

class UserState(DataModel):
    name: str = field()
    email: str = field()
    active: bool = field(initial=True)
    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

Step 3: Define Aggregate​

Create aggregate.py:

from fluvius.domain import Aggregate, action
from .state import UserState

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):
        # Business logic
        if not email or '@' not in email:
            raise ValueError("Invalid email")
        
        self._state = UserState(name=name, email=email, active=True)
        return self._state
    
    @action(evt_key='user-updated', resources=['user'])
    async def update_user(self, name: str = None, email: str = None):
        if self._state is None:
            raise ValueError("User not created")
        
        updates = {}
        if name is not None:
            updates['name'] = name
        if email is not None:
            if '@' not in email:
                raise ValueError("Invalid email")
            updates['email'] = email
        
        updates['updated_at'] = datetime.now()
        self._state = self._state.update(**updates)
        return self._state
    
    @action(evt_key='user-deactivated', resources=['user'])
    async def deactivate_user(self):
        if self._state is None:
            raise ValueError("User not created")
        
        self._state = self._state.update(active=False, updated_at=datetime.now())
        return self._state

Step 4: Define Domain​

Create domain.py:

from fluvius.domain import Domain
from .aggregate import UserAggregate

class UserDomain(Domain):
    __aggregate__ = UserAggregate
    
    class Meta:
        revision = 1
        tags = ["user", "identity"]
        title = "User Management Domain"
        description = "Domain for managing user accounts"

Step 5: Export Domain​

Create __init__.py:

from .domain import UserDomain
from .aggregate import UserAggregate
from .state import UserState

__all__ = ['UserDomain', 'UserAggregate', 'UserState']

Using the Domain​

Register with FastAPI​

from fluvius.fastapi import FluviusFastAPI
from myapp.domains.user import UserDomain

app = FluviusFastAPI()
app.register_domain(UserDomain)

Use Domain Directly​

from fluvius.domain.context import SanicContext
from myapp.domains.user import UserDomain
from fluvius.data import UUID_GENR

# Create context
ctx = SanicContext.create(namespace='app-user')

# Create domain
domain = UserDomain(ctx)

# Set aggregate root
user_id = UUID_GENR()
domain.set_aggroot('user', user_id)

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

# Process command
response = await domain.process_command(command)

Complete Example​

File Structure​

myapp/
└── domains/
    └── user/
        ├── __init__.py
        ├── domain.py
        ├── aggregate.py
        └── state.py

Complete Code​

state.py:

from fluvius.data import DataModel, field
from datetime import datetime

class UserState(DataModel):
    name: str = field()
    email: str = field()
    active: bool = field(initial=True)
    created_at: datetime = field(default_factory=datetime.now)

aggregate.py:

from fluvius.domain import Aggregate, action
from .state import UserState

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

domain.py:

from fluvius.domain import Domain
from .aggregate import UserAggregate

class UserDomain(Domain):
    __aggregate__ = UserAggregate

init.py:

from .domain import UserDomain
from .aggregate import UserAggregate
from .state import UserState

__all__ = ['UserDomain', 'UserAggregate', 'UserState']

Best Practices​

1. Keep Domains Focused​

Each domain should have a single responsibility:

# Good: Focused domain
class UserDomain(Domain):
    # User management only
    pass

# Bad: Too many responsibilities
class UserOrderPaymentDomain(Domain):
    # Too many concerns
    pass

2. Use Meaningful Names​

Use clear, descriptive names:

# Good: Clear names
@action(evt_key='user-account-activated')

# Bad: Unclear names
@action(evt_key='user-updated')

3. Validate in Aggregates​

Enforce business rules in aggregates:

@action(evt_key='user-created')
async def create_user(self, name: str, email: str):
    # Validate business rules
    if not email or '@' not in email:
        raise ValueError("Invalid email")
    pass

4. Use Type Hints​

Always use type hints:

@action(evt_key='user-created')
async def create_user(self, name: str, email: str) -> dict:
    return {'name': name, 'email': email}

Next Steps​