Skip to main content

Design Patterns

info

Fluvius Framework implements several proven design patterns to help you build maintainable applications.

Domain-Driven Design (DDD)​

Aggregate Pattern​

Purpose: Encapsulate business logic and enforce invariants.

Implementation:

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 and invariant enforcement
        if not email or '@' not in email:
            raise ValueError("Invalid email")
        return {'name': name, 'email': email}

Benefits:

  • Encapsulates business logic
  • Enforces invariants
  • Clear boundaries
  • Testable in isolation

Bounded Context Pattern​

Purpose: Define clear boundaries between domains.

Implementation:

# User domain (bounded context)
class UserDomain(Domain):
    __aggregate__ = UserAggregate
    __namespace__ = 'app-user'

# Order domain (different bounded context)
class OrderDomain(Domain):
    __aggregate__ = OrderAggregate
    __namespace__ = 'app-order'

Benefits:

  • Clear domain boundaries
  • Independent evolution
  • Reduced coupling
  • Team autonomy

Ubiquitous Language Pattern​

Purpose: Use business terminology in code.

Implementation:

# Good: Uses business language
@action(evt_key='order-placed')
async def place_order(self, items: list):
    pass

# Bad: Uses technical terms
@action(evt_key='order-created')
async def create_order(self, products: list):
    pass

Benefits:

  • Code reflects business
  • Better communication
  • Less translation needed
  • Clearer intent

Command Query Responsibility Segregation (CQRS)​

Command Pattern​

Purpose: Separate commands (writes) from queries (reads).

Implementation:

# Command (write)
command = domain.create_command('create-user', {
    'name': 'John Doe',
    'email': '[email protected]'
})
response = await domain.process_command(command)

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

Benefits:

  • Independent scaling
  • Optimized models
  • Clear separation
  • Better performance

Command Handler Pattern​

Purpose: Handle commands through aggregates.

Implementation:

class UserAggregate(Aggregate):
    @action(evt_key='user-created')
    async def create_user(self, name: str, email: str):
        # Command handler logic
        return {'name': name, 'email': email}

Benefits:

  • Centralized command handling
  • Business logic in one place
  • Easy to test
  • Clear flow

Event Sourcing​

Event Store Pattern​

Purpose: Store all changes as events.

Implementation:

@action(evt_key='user-created', resources=['user'])
async def create_user(self, name: str, email: str):
    # Event automatically stored in event store
    return {'name': name, 'email': email}

Benefits:

  • Complete audit trail
  • Event replay
  • Time travel
  • Event-driven integration

Event Sourcing Pattern​

Purpose: Rebuild state from events.

Implementation:

# State is rebuilt from events
# Events: [user-created, user-updated, user-deactivated]
# Current state: {name: 'John', email: '[email protected]', active: False}

Benefits:

  • State reconstruction
  • Historical queries
  • Debugging
  • Audit compliance

Repository Pattern​

State Manager Pattern​

Purpose: Abstract data access.

Implementation:

# Abstract interface
user = await domain.statemgr.fetch('user', user_id)
users = await domain.statemgr.find('user', active=True)

Benefits:

  • Abstract data access
  • Testable
  • Swappable implementations
  • Clean separation

Factory Pattern​

Domain Factory​

Purpose: Create domain instances.

Implementation:

from fluvius.domain.context import SanicContext

ctx = SanicContext.create(namespace='app-user')
domain = UserDomain(ctx)

Benefits:

  • Centralized creation
  • Consistent configuration
  • Dependency injection
  • Easy testing

Command Factory​

Purpose: Create commands.

Implementation:

command = domain.create_command('create-user', {
    'name': 'John Doe',
    'email': '[email protected]'
})

Benefits:

  • Type-safe creation
  • Validation
  • Consistent structure
  • Easy to use

Strategy Pattern​

Driver Strategy​

Purpose: Swappable database drivers.

Implementation:

from fluvius.data import PostgreSQLDriver, MongoDBDriver

# Use PostgreSQL
driver = PostgreSQLDriver(connection_string)

# Or MongoDB
driver = MongoDBDriver(connection_string)

Benefits:

  • Swappable implementations
  • Database agnostic
  • Easy testing
  • Flexibility

Storage Strategy​

Purpose: Swappable storage backends.

Implementation:

from fluvius.media import LocalStorage, S3Storage

# Use local storage
storage = LocalStorage(path='/uploads')

# Or S3
storage = S3Storage(bucket='my-bucket')

Benefits:

  • Multiple backends
  • Easy switching
  • Cloud-ready
  • Testable

Observer Pattern​

Event Handlers​

Purpose: React to events.

Implementation:

@event_handler('user-created')
async def send_welcome_email(event):
    # React to user-created event
    await email_service.send(event.payload['email'])

Benefits:

  • Loose coupling
  • Event-driven
  • Extensible
  • Testable

Middleware Pattern​

Request Middleware​

Purpose: Process requests/responses.

Implementation:

from fluvius.fastapi import middleware

@middleware
async def logging_middleware(request, call_next):
    # Log request
    response = await call_next(request)
    # Log response
    return response

Benefits:

  • Cross-cutting concerns
  • Reusable logic
  • Chainable
  • Testable

Dependency Injection​

Context Injection​

Purpose: Inject dependencies through context.

Implementation:

from fluvius.domain.context import SanicContext

ctx = SanicContext.create(namespace='app-user')
domain = UserDomain(ctx)  # Context injected

Benefits:

  • Loose coupling
  • Testable
  • Flexible
  • Maintainable

Template Method Pattern​

Aggregate Template​

Purpose: Define aggregate structure.

Implementation:

class Aggregate:
    async def process(self, command):
        # Template method
        self.validate(command)
        result = await self.execute(command)
        self.generate_event(result)
        return result

Benefits:

  • Consistent structure
  • Reusable logic
  • Extensible
  • Clear flow

Builder Pattern​

Query Builder​

Purpose: Build complex queries.

Implementation:

from fluvius.query import QueryBuilder

query = QueryBuilder('user')
    .filter(active=True)
    .sort('created_at', desc=True)
    .limit(10)

results = await query.execute()

Benefits:

  • Fluent interface
  • Composable
  • Readable
  • Flexible

Decorator Pattern​

Action Decorator​

Purpose: Add behavior to methods.

Implementation:

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

Benefits:

  • Non-invasive
  • Reusable
  • Composable
  • Clear intent

Singleton Pattern​

Context Singleton​

Purpose: Single context instance per request.

Implementation:

ctx = SanicContext.create(namespace='app-user')
# Single instance per namespace

Benefits:

  • Single instance
  • Shared state
  • Resource efficiency
  • Consistent access

Pattern Combinations​

DDD + CQRS + Event Sourcing​

Fluvius combines these patterns:

# DDD: Aggregate with business logic
class UserAggregate(Aggregate):
    # CQRS: Command handler
    @action(evt_key='user-created')
    async def create_user(self, name: str, email: str):
        # Event Sourcing: Event generated
        return {'name': name, 'email': email}

Benefits:

  • Best of all patterns
  • Scalable architecture
  • Maintainable code
  • Proven patterns

Best Practices​

1. Use Patterns Appropriately​

Don't over-engineer. Use patterns when they add value.

2. Keep Patterns Simple​

Simple implementations are better than complex ones.

3. Document Patterns​

Document which patterns you use and why.

4. Test Patterns​

Test pattern implementations thoroughly.

Next Steps​