Skip to main content
Unlisted page
This page is unlisted. Search engines will not index it, and only users having a direct link can access it.

Module Structure

info

Fluvius Framework is organized into independent modules that can be used together or separately.

Module Overview​

Fluvius consists of several modules, each providing specific functionality:

fluvius/
├── domain/          # Core domain logic (CQRS, Event Sourcing)
├── fastapi/         # FastAPI integration
├── data/            # Data layer abstraction
├── worker/          # Background job processing
├── query/           # Query builder
├── media/           # Media/file management
├── casbin/          # Authorization (Casbin integration)
├── nava/            # Navigation/routing
└── manager/         # Application management

Domain Module​

Purpose: Core domain logic with CQRS and event sourcing.

Key Components:

  • Domain: Main domain class
  • Aggregate: Business logic container
  • Command: Command representation
  • Event: Event representation
  • StateManager: State query interface

Dependencies:

  • data module (for data models)

Usage:

from fluvius.domain import Domain, Aggregate, action

FastAPI Module​

Purpose: FastAPI integration for REST APIs.

Key Components:

  • FluviusFastAPI: FastAPI application wrapper
  • DomainRouter: Automatic endpoint generation
  • Middleware: Request/response middleware

Dependencies:

  • domain module
  • FastAPI framework

Usage:

from fluvius.fastapi import FluviusFastAPI

Data Module​

Purpose: Data layer abstraction and models.

Key Components:

  • DataModel: Base data model class
  • Driver: Database driver interface
  • PostgreSQLDriver: PostgreSQL implementation
  • MongoDBDriver: MongoDB implementation
  • SQLiteDriver: SQLite implementation

Dependencies:

  • None (core module)

Usage:

from fluvius.data import DataModel, field, PostgreSQLDriver

Worker Module​

Purpose: Background job processing.

Key Components:

  • Worker: Worker instance
  • task: Task decorator
  • Scheduler: Task scheduler

Dependencies:

  • ARQ (Redis-based job queue)

Usage:

from fluvius.worker import Worker, task

Query Module​

Purpose: Flexible query builder.

Key Components:

  • QueryBuilder: Query builder class
  • Filter: Filter expressions
  • Sort: Sort expressions

Dependencies:

  • data module

Usage:

from fluvius.query import QueryBuilder

Media Module​

Purpose: File upload and storage.

Key Components:

  • MediaManager: Media manager
  • StorageBackend: Storage backend interface
  • LocalStorage: Local file storage
  • S3Storage: AWS S3 storage

Dependencies:

  • None (optional S3 support)

Usage:

from fluvius.media import MediaManager

Casbin Module​

Purpose: Policy-based authorization.

Key Components:

  • CasbinEnforcer: Casbin enforcer wrapper
  • PolicyAdapter: Policy storage adapter

Dependencies:

  • Casbin library

Usage:

from fluvius.casbin import CasbinEnforcer

Purpose: Navigation and routing utilities.

Key Components:

  • Navigation helpers
  • Route generation

Dependencies:

  • domain module

Manager Module​

Purpose: Application management and utilities.

Key Components:

  • Application lifecycle
  • Configuration management
  • Service registration

Dependencies:

  • Multiple modules

Module Dependencies​

domain (core)
  │
  ├──► data (required)
  │
  └──► casbin (optional)

fastapi
  │
  └──► domain (required)

worker
  │
  └──► domain (optional)

query
  │
  └──► data (required)

media
  │
  └──► (standalone)

casbin
  │
  └──► (standalone)

nava
  │
  └──► domain (required)

manager
  │
  └──► (depends on multiple)

Installing Modules​

Install All Modules​

pip install fluvius

Install Specific Modules​

# Core domain module
pip install fluvius-domain

# FastAPI integration
pip install fluvius-fastapi

# Worker support
pip install fluvius-worker

Module Configuration​

Domain Module​

from fluvius.domain import Domain

class UserDomain(Domain):
    __aggregate__ = UserAggregate
    __namespace__ = 'app-user'

FastAPI Module​

from fluvius.fastapi import FluviusFastAPI

app = FluviusFastAPI()
app.register_domain(UserDomain)

Data Module​

from fluvius.data import PostgreSQLDriver

driver = PostgreSQLDriver(
    connection_string="postgresql://user:pass@localhost/db"
)

Worker Module​

from fluvius.worker import Worker

worker = Worker(redis_url="redis://localhost:6379")

Module Extensibility​

Custom Drivers​

Create custom database drivers:

from fluvius.data import Driver

class CustomDriver(Driver):
    async def connect(self):
        # Custom connection logic
        pass
    
    async def execute(self, query, params):
        # Custom execution logic
        pass

Custom Storage Backends​

Create custom storage backends:

from fluvius.media import StorageBackend

class CustomStorage(StorageBackend):
    async def upload(self, file, path):
        # Custom upload logic
        pass

Module Best Practices​

1. Use Only What You Need​

Don't install modules you don't use:

# Good: Only import what you need
from fluvius.domain import Domain
from fluvius.data import DataModel

# Bad: Import everything
from fluvius import *

2. Keep Modules Independent​

Design domains to work independently:

# Good: Domain doesn't depend on FastAPI
class UserDomain(Domain):
    __aggregate__ = UserAggregate

# Bad: Domain depends on FastAPI
class UserDomain(Domain):
    def __init__(self, fastapi_app):
        # Tight coupling
        pass

3. Use Module Interfaces​

Use interfaces, not implementations:

# Good: Use Driver interface
from fluvius.data import Driver

def process_data(driver: Driver):
    # Works with any driver
    pass

# Bad: Use specific implementation
from fluvius.data import PostgreSQLDriver

def process_data(driver: PostgreSQLDriver):
    # Tightly coupled
    pass

Next Steps​