Skip to main content

Quick Start

info

This guide will help you build your first Fluvius application in minutes.

Create Your First Domain​

Let's create a simple user management domain to get started.

Step 1: Define Your Aggregate​

Create a file user_aggregate.py:

from fluvius.domain import Aggregate, action
from fluvius.data import DataModel, field

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

class UserAggregate(Aggregate):
    def __init__(self, domain):
        super().__init__(domain)
        self._state = UserState()
    
    @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
    
    @action(evt_key='user-updated', resources=['user'])
    async def update_user(self, name: str = None, email: str = None):
        updates = {}
        if name is not None:
            updates['name'] = name
        if email is not None:
            updates['email'] = email
        self._state = self._state.update(**updates)
        return self._state
    
    @action(evt_key='user-deactivated', resources=['user'])
    async def deactivate_user(self):
        self._state = self._state.update(active=False)
        return self._state

Step 2: Create Your Domain​

Create a file user_domain.py:

from fluvius.domain import Domain
from .user_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 3: Process Commands​

Create a file main.py:

import asyncio
from fluvius.domain.context import SanicContext
from fluvius.data import UUID_GENR
from user_domain import UserDomain

async def main():
    # Create context
    ctx = SanicContext.create(namespace='app-user')
    
    # Create domain instance
    domain = UserDomain(ctx)
    
    # Set aggregate root
    user_id = UUID_GENR()
    domain.set_aggroot('user', user_id)
    
    # Create and process commands
    commands = [
        domain.create_command('create-user', {
            'name': 'John Doe',
            'email': '[email protected]'
        }),
        domain.create_command('update-user', {
            'name': 'Jane Doe'
        }),
        domain.create_command('deactivate-user')
    ]
    
    # Process commands
    async for response in domain.command_processor.process(*commands):
        print(f'Response: {response}')

if __name__ == '__main__':
    asyncio.run(main())

Step 4: Run Your Application​

python main.py

What Happened?​

  1. Aggregate: Defined business logic and state management
  2. Domain: Wired the aggregate to the domain framework
  3. Commands: Created commands to modify state
  4. Events: Commands automatically generated events
  5. Processing: Domain processed commands and returned responses

Next Steps​