Skip to main content

Configuration

info

Configure Fluvius Framework to match your application's needs.

Basic Configuration​

Environment Variables​

Create a .env file:

# Database
DATABASE_URL=postgresql://user:password@localhost:5432/mydb

# Redis (for workers)
REDIS_URL=redis://localhost:6379/0

# Application
SECRET_KEY=your-secret-key-here
DEBUG=False
LOG_LEVEL=INFO

# Namespace
NAMESPACE=myapp

Settings Class​

Create a settings file:

# config/settings.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # Database
    database_url: str
    redis_url: str = "redis://localhost:6379/0"
    
    # Application
    secret_key: str
    debug: bool = False
    log_level: str = "INFO"
    
    # Namespace
    namespace: str = "myapp"
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

Domain Configuration​

Domain Meta​

Configure domain metadata:

from fluvius.domain import Domain

class UserDomain(Domain):
    __aggregate__ = UserAggregate
    
    class Meta:
        revision = 1
        tags = ["user", "identity"]
        title = "User Management Domain"
        description = "Domain for managing user accounts"

Namespace Configuration​

Set namespace for isolation:

from fluvius.domain.context import SanicContext

# Create context with namespace
ctx = SanicContext.create(namespace='app-user')
domain = UserDomain(ctx)

Database Configuration​

PostgreSQL​

from fluvius.data import PostgreSQLDriver

driver = PostgreSQLDriver(
    connection_string="postgresql://user:pass@localhost/db",
    pool_size=10,
    max_overflow=20
)

MongoDB​

from fluvius.data import MongoDBDriver

driver = MongoDBDriver(
    connection_string="mongodb://localhost:27017/",
    database="mydb"
)

SQLite​

from fluvius.data import SQLiteDriver

driver = SQLiteDriver(
    database_path=":memory:",  # or "path/to/database.db"
    check_same_thread=False
)

FastAPI Configuration​

Basic App​

from fluvius.fastapi import FluviusFastAPI

app = FluviusFastAPI(
    title="My API",
    version="1.0.0",
    description="My Fluvius API"
)

With CORS​

from fastapi.middleware.cors import CORSMiddleware
from fluvius.fastapi import FluviusFastAPI

app = FluviusFastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

With Authentication​

from fluvius.fastapi import FluviusFastAPI
from fluvius.fastapi.authentication import AuthMiddleware

app = FluviusFastAPI()

app.add_middleware(AuthMiddleware)

Worker Configuration​

Basic Worker​

from fluvius.worker import Worker

worker = Worker(
    redis_url="redis://localhost:6379/0",
    max_jobs=10
)

With Settings​

from fluvius.worker import Worker
from myapp.config.settings import settings

worker = Worker(
    redis_url=settings.redis_url,
    max_jobs=settings.max_worker_jobs
)

Logging Configuration​

Basic Logging​

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

Structured Logging​

import logging
import json

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
        }
        if hasattr(record, 'extra'):
            log_data.update(record.extra)
        return json.dumps(log_data)

handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logging.root.addHandler(handler)
logging.root.setLevel(logging.INFO)

Media Configuration​

Local Storage​

from fluvius.media import MediaManager, LocalStorage

storage = LocalStorage(path="/uploads")
media = MediaManager(storage=storage)

S3 Storage​

from fluvius.media import MediaManager, S3Storage

storage = S3Storage(
    bucket="my-bucket",
    region="us-east-1",
    access_key_id="...",
    secret_access_key="..."
)
media = MediaManager(storage=storage)

Casbin Configuration​

Basic Enforcer​

from fluvius.casbin import CasbinEnforcer

enforcer = CasbinEnforcer(
    model_path="path/to/model.conf",
    policy_path="path/to/policy.csv"
)

With Database Adapter​

from fluvius.casbin import CasbinEnforcer
from casbin_sqlalchemy_adapter import Adapter

adapter = Adapter(engine)
enforcer = CasbinEnforcer(
    model_path="path/to/model.conf",
    adapter=adapter
)

Configuration Files​

config.ini​

[database]
url = postgresql://user:pass@localhost/db
pool_size = 10
max_overflow = 20

[redis]
url = redis://localhost:6379/0

[application]
secret_key = your-secret-key
debug = False
log_level = INFO
namespace = myapp

pyproject.toml​

[tool.fluvius]
namespace = "myapp"
database_url = "postgresql://user:pass@localhost/db"
redis_url = "redis://localhost:6379/0"

[tool.fluvius.logging]
level = "INFO"
format = "json"

Environment-Specific Configuration​

Development​

# config/development.py
from .settings import Settings

class DevelopmentSettings(Settings):
    debug: bool = True
    log_level: str = "DEBUG"
    database_url: str = "postgresql://localhost/devdb"

Production​

# config/production.py
from .settings import Settings

class ProductionSettings(Settings):
    debug: bool = False
    log_level: str = "WARNING"
    database_url: str = os.getenv("DATABASE_URL")

Testing​

# config/testing.py
from .settings import Settings

class TestingSettings(Settings):
    debug: bool = True
    log_level: str = "DEBUG"
    database_url: str = "sqlite:///:memory:"

Configuration Best Practices​

1. Use Environment Variables​

Store secrets in environment variables:

import os

database_url = os.getenv("DATABASE_URL")
secret_key = os.getenv("SECRET_KEY")

2. Validate Configuration​

Use Pydantic for validation:

from pydantic import BaseSettings, validator

class Settings(BaseSettings):
    database_url: str
    
    @validator('database_url')
    def validate_database_url(cls, v):
        if not v.startswith(('postgresql://', 'mongodb://', 'sqlite://')):
            raise ValueError('Invalid database URL')
        return v

3. Use Defaults​

Provide sensible defaults:

class Settings(BaseSettings):
    log_level: str = "INFO"
    debug: bool = False
    pool_size: int = 10

4. Separate Secrets​

Keep secrets separate:

# .env (gitignored)
SECRET_KEY=actual-secret-key
DATABASE_PASSWORD=actual-password
# .env.example (committed)
SECRET_KEY=your-secret-key-here
DATABASE_PASSWORD=your-password-here

5. Document Configuration​

Document all configuration options:

class Settings(BaseSettings):
    """Application settings.
    
    Attributes:
        database_url: Database connection string
        redis_url: Redis connection string
        secret_key: Secret key for encryption
    """
    database_url: str
    redis_url: str
    secret_key: str

Next Steps​