Skip to main content

Key Features

info

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
  • MongoDB: Document database support
  • SQLite: Development and testing
  • Extensible: Easy to add new backends
from fluvius.data import PostgreSQLDriver, MongoDBDriver

# Use PostgreSQL
driver = PostgreSQLDriver(connection_string)

# Or MongoDB
driver = MongoDBDriver(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​

FeatureFluviusDjangoFastAPIEventStore
DDD Support✅ Built-in❌❌❌
Event Sourcing✅ Built-in❌❌✅
CQRS✅ Built-in❌❌✅
FastAPI Integration✅❌✅❌
Multiple DBs✅✅❌❌
Type Safety✅Partial✅❌
Python Native✅✅✅❌

Next Steps​