Register Services

thông tin

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

• Learn about Custom Middleware
• Explore Error Handling
• Check New Module