Skip to main content

Register Services

info

Learn how to register and use services in Fluvius Framework.

Service Registration​

Register Domain​

Register domains with 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)

Register with Context​

from fluvius.domain.context import SanicContext

ctx = SanicContext.create(namespace='app-user')
domain = UserDomain(ctx)
app.register_domain(domain)

Dependency Injection​

FastAPI Dependencies​

Use FastAPI dependencies:

from fastapi import Depends

def get_domain():
    ctx = SanicContext.create(namespace='app-user')
    return UserDomain(ctx)

@app.post("/api/users")
async def create_user(domain: UserDomain = Depends(get_domain)):
    command = domain.create_command('create-user', {...})
    return await domain.process_command(command)

Service Locator​

class ServiceLocator:
    def __init__(self):
        self._services = {}
    
    def register(self, name: str, service):
        self._services[name] = service
    
    def get(self, name: str):
        return self._services.get(name)

locator = ServiceLocator()
locator.register('user_domain', UserDomain(ctx))

Service Lifecycle​

Startup Events​

@app.on_event("startup")
async def startup():
    # Initialize services
    await database.connect()
    await redis.connect()

@app.on_event("shutdown")
async def shutdown():
    # Cleanup services
    await database.disconnect()
    await redis.disconnect()

Service Configuration​

Environment-Based Services​

import os

def get_database_url():
    env = os.getenv("ENVIRONMENT", "development")
    if env == "production":
        return os.getenv("DATABASE_URL")
    else:
        return "sqlite:///:memory:"

driver = PostgreSQLDriver(connection_string=get_database_url())

Next Steps​