Folder Structure
info
A well-organized folder structure helps maintain clean, scalable applications.
Recommended Structure
my-fluvius-app/
├── src/
│ └── myapp/
│ ├── __init__.py
│ ├── domains/
│ │ ├── __init__.py
│ │ ├── user/
│ │ │ ├── __init__.py
│ │ │ ├── domain.py
│ │ │ ├── aggregate.py
│ │ │ ├── state.py
│ │ │ └── commands.py
│ │ └── order/
│ │ ├── __init__.py
│ │ ├── domain.py
│ │ ├── aggregate.py
│ │ ├── state.py
│ │ └── commands.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── main.py
│ │ ├── routes.py
│ │ └── middleware.py
│ ├── config/
│ │ ├── __init__.py
│ │ ├── settings.py
│ │ └── database.py
│ └── utils/
│ ├── __init__.py
│ └── helpers.py
├── tests/
│ ├── __init__.py
│ ├── unit/
│ │ ├── test_user_aggregate.py
│ │ └── test_order_aggregate.py
│ └── integration/
│ ├── test_user_api.py
│ └── test_order_api.py
├── alembic/
│ ├── versions/
│ └── env.py
├── requirements.txt
├── pyproject.toml
├── .env
├── .gitignore
└── README.md
Domain Structure
Single Domain
domains/
└── user/
├── __init__.py # Domain exports
├── domain.py # Domain class
├── aggregate.py # Aggregate class
├── state.py # State models
├── commands.py # Command definitions (optional)
└── events.py # Event definitions (optional)
Multiple Domains
domains/
├── user/
│ ├── __init__.py
│ ├── domain.py
│ ├── aggregate.py
│ └── state.py
├── order/
│ ├── __init__.py
│ ├── domain.py
│ ├── aggregate.py
│ └── state.py
└── payment/
├── __init__.py
├── domain.py
├── aggregate.py
└── state.py
Domain File Contents
domain.py
from fluvius.domain import Domain
from .aggregate import UserAggregate
class UserDomain(Domain):
__aggregate__ = UserAggregate
class Meta:
revision = 1
tags = ["user", "identity"]
title = "User Management Domain"
description = "Domain for managing user accounts"
aggregate.py
from fluvius.domain import Aggregate, action
from .state import UserState
class UserAggregate(Aggregate):
def __init__(self, domain):
super().__init__(domain)
self._state = None
@action(evt_key='user-created', resources=['user'])
async def create_user(self, name: str, email: str):
self._state = UserState(name=name, email=email, active=True)
return self._state
@action(evt_key='user-updated', resources=['user'])
async def update_user(self, name: str = None, email: str = None):
updates = {}
if name is not None:
updates['name'] = name
if email is not None:
updates['email'] = email
self._state = self._state.update(**updates)
return self._state
state.py
from fluvius.data import DataModel, field
from datetime import datetime
class UserState(DataModel):
name: str = field()
email: str = field()
active: bool = field(initial=True)
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
API Structure
main.py
from fastapi import FastAPI
from fluvius.fastapi import FluviusFastAPI
from myapp.domains.user import UserDomain
from myapp.domains.order import OrderDomain
app = FluviusFastAPI()
# Register domains
app.register_domain(UserDomain)
app.register_domain(OrderDomain)
@app.get("/")
async def root():
return {"message": "Hello World"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
routes.py (Optional)
from fastapi import APIRouter
from myapp.domains.user import UserDomain
router = APIRouter()
@router.get("/users/{user_id}")
async def get_user(user_id: str):
# Custom route logic
pass
Configuration Structure
settings.py
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
redis_url: str
secret_key: str
class Config:
env_file = ".env"
settings = Settings()
database.py
from fluvius.data import PostgreSQLDriver
from myapp.config.settings import settings
driver = PostgreSQLDriver(connection_string=settings.database_url)
Test Structure
Unit Tests
tests/
└── unit/
├── test_user_aggregate.py
└── test_order_aggregate.py
Example:
import pytest
from myapp.domains.user import UserAggregate, UserDomain
from fluvius.domain.context import SanicContext
@pytest.mark.asyncio
async def test_create_user():
ctx = SanicContext.create(namespace='test-user')
domain = UserDomain(ctx)
aggregate = UserAggregate(domain)
result = await aggregate.create_user('John', 'john@example.com')
assert result['name'] == 'John'
Integration Tests
tests/
└── integration/
├── test_user_api.py
└── test_order_api.py
Alternative Structures
Flat Structure (Small Projects)
myapp/
├── __init__.py
├── user_domain.py
├── user_aggregate.py
├── order_domain.py
├── order_aggregate.py
└── main.py
Feature-Based Structure
myapp/
├── features/
│ ├── user_management/
│ │ ├── domain.py
│ │ ├── aggregate.py
│ │ └── api.py
│ └── order_processing/
│ ├── domain.py
│ ├── aggregate.py
│ └── api.py
Best Practices
1. One Domain Per Folder
Keep each domain in its own folder:
domains/
└── user/ # One domain
├── domain.py
└── aggregate.py
2. Separate State Models
Keep state models separate:
user/
├── aggregate.py # Business logic
└── state.py # State models
3. Group Related Files
Group related files together:
user/
├── domain.py # Domain definition
├── aggregate.py # Business logic
├── state.py # State models
└── commands.py # Command helpers (optional)
4. Use init.py
Export from __init__.py:
# domains/user/__init__.py
from .domain import UserDomain
from .aggregate import UserAggregate
from .state import UserState
__all__ = ['UserDomain', 'UserAggregate', 'UserState']
5. Keep Tests Close
Keep tests near code (optional):
myapp/
├── domains/
│ └── user/
│ ├── domain.py
│ └── test_domain.py # Test in same folder
Or separate:
tests/
└── domains/
└── user/
└── test_domain.py
Migration Structure
Alembic Structure
alembic/
├── versions/
│ ├── 001_initial.py
│ └── 002_add_user_table.py
├── env.py
└── script.py.mako
Environment Files
.env
DATABASE_URL=postgresql://user:pass@localhost/db
REDIS_URL=redis://localhost:6379
SECRET_KEY=your-secret-key
.env.example
DATABASE_URL=postgresql://user:pass@localhost/db
REDIS_URL=redis://localhost:6379
SECRET_KEY=your-secret-key
Next Steps
- Learn about Configuration
- Check Quick Start
- Explore Usage Guide