Design Patterns

thông tin

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': 'john@example.com'
})
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: 'john@example.com', 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': 'john@example.com'
})

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

• Learn about Core Components
• Explore Module Structure
• Check Getting Started