Key Features

thông tin

Fluvius Framework provides a comprehensive set of features for building domain-driven, event-sourced applications.

Core Features

1. Domain-Driven Design (DDD)

Build applications around your business domain:

• Aggregates: Encapsulate business logic and enforce invariants
• Bounded Contexts: Clear domain boundaries
• Ubiquitous Language: Code reflects business terminology
• Domain Events: Express business occurrences

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

2. Command Query Responsibility Segregation (CQRS)

Separate read and write operations:

• Commands: Modify state through aggregates
• Queries: Read from optimized state stores
• Independent Scaling: Scale reads and writes separately
• Optimized Performance: Different models for reads and writes

# Command (write)
command = domain.create_command('create-user', {...})
response = await domain.process_command(command)

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

3. Event Sourcing

Track all changes as immutable events:

• Event Log: Complete history of all changes
• Event Replay: Rebuild state from events
• Time Travel: Query state at any point in time
• Audit Trail: Complete change history

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

4. FastAPI Integration

Build REST APIs quickly:

• Automatic Endpoints: Generate endpoints from domains
• Request Validation: Pydantic-based validation
• OpenAPI Documentation: Auto-generated API docs
• Async Support: Full async/await support

from fluvius.fastapi import FluviusFastAPI

app = FluviusFastAPI()
app.register_domain(UserDomain)
# Endpoints automatically created

5. Multiple Database Backends

Support for various databases:

• PostgreSQL: Production-ready relational database
• SQLite: Development and testing
• Extensible: Easy to add new backends

from fluvius.data import PostgreSQLDriver

# Use PostgreSQL
driver = PostgreSQLDriver(connection_string)

6. State Management

Built-in state management:

• Current State: Fast access to current state
• Queries: Filter, sort, paginate
• Transactions: ACID-compliant operations
• Caching: Optional caching layer

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

# Query with filters
active_users = await domain.statemgr.find('user', active=True)

# Pagination
users = await domain.statemgr.find('user', limit=10, offset=0)

7. Background Workers

Async job processing:

• ARQ Integration: Redis-based job queue
• Task Scheduling: Cron-like scheduling
• Retry Logic: Automatic retries with backoff
• Monitoring: Job status tracking

from fluvius.worker import Worker, task

@task
async def process_payment(payment_id: str):
    # Background processing
    pass

8. Media Management

File upload and storage:

• Multiple Storage Backends: Local, S3, etc.
• File Validation: Type and size validation
• Metadata Storage: Store file metadata
• URL Generation: Generate access URLs

from fluvius.media import MediaManager

media = MediaManager()
file_info = await media.upload(file, filename='document.pdf')

9. Query Builder

Flexible querying:

• Filtering: Complex filter expressions
• Sorting: Multi-field sorting
• Pagination: Cursor and offset-based
• Aggregations: Count, sum, average

from fluvius.query import QueryBuilder

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

results = await query.execute()

10. Authorization (Casbin)

Policy-based access control:

• RBAC: Role-based access control
• ABAC: Attribute-based access control
• Policy Enforcement: Automatic policy checking
• Integration: Built-in Casbin integration

from fluvius.casbin import CasbinEnforcer

enforcer = CasbinEnforcer()
# Policies automatically enforced

11. Type Safety

Full type hint support:

• Type Checking: mypy compatible
• IDE Support: Better autocomplete
• Runtime Validation: Pydantic validation
• Documentation: Self-documenting code

from fluvius.domain import Aggregate, action
from typing import Optional

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

12. Testing Support

Comprehensive testing utilities:

• Test Fixtures: Domain testing helpers
• Mocking: Easy mocking of dependencies
• Event Assertions: Test event generation
• State Assertions: Verify state changes

from fluvius.testing import DomainTestCase

class TestUserDomain(DomainTestCase):
    async def test_create_user(self):
        response = await self.domain.process_command(...)
        self.assertEventGenerated('user-created')

13. Logging & Monitoring

Built-in observability:

• Structured Logging: JSON logging support
• Event Logging: Automatic event logging
• Metrics: Performance metrics
• Tracing: Request tracing support

14. Extensibility

Easy to extend:

• Custom Drivers: Add new database backends
• Middleware: Custom request/response middleware
• Plugins: Plugin system for extensions
• Hooks: Lifecycle hooks

Feature Comparison

Feature	Fluvius	Django	FastAPI	EventStore
DDD Support	✅ Built-in	❌	❌	❌
Event Sourcing	✅ Built-in	❌	❌	✅
CQRS	✅ Built-in	❌	❌	✅
FastAPI Integration	✅	❌	✅	❌
Multiple DBs	✅	✅	❌	❌
Type Safety	✅	Partial	✅	❌
Python Native	✅	✅	✅	❌

Next Steps

• Learn about Target Audience
• Explore Core Principles
• Start building with Getting Started