Skip to main content

Error Handling

info

Learn how to handle errors effectively in Fluvius Framework.

Aggregate Error Handling​

Validation Errors​

from fluvius.domain import Aggregate, action

class UserAggregate(Aggregate):
    @action(evt_key='user-created')
    async def create_user(self, name: str, email: str):
        # Validate input
        if not name or len(name) < 2:
            raise ValueError("Name must be at least 2 characters")
        
        if not email or '@' not in email:
            raise ValueError("Invalid email address")
        
        # Business logic
        pass

Business Rule Errors​

@action(evt_key='account-withdrawn')
async def withdraw(self, amount: float):
    # Check business rules
    if self._state.balance < amount:
        raise ValueError("Insufficient funds")
    
    if amount <= 0:
        raise ValueError("Amount must be positive")
    
    # Process withdrawal
    pass

Command Processing Errors​

Error Handling​

try:
    command = domain.create_command('create-user', {...})
    response = await domain.process_command(command)
except ValueError as e:
    # Handle validation error
    logger.error(f"Validation error: {e}")
    return {"error": str(e)}
except Exception as e:
    # Handle other errors
    logger.exception("Command processing error")
    return {"error": "Internal error"}

API Error Handling​

HTTP Exception Handling​

from fastapi import HTTPException

@app.post("/api/users")
async def create_user(request: CreateUserRequest):
    try:
        command = domain.create_command('create-user', {...})
        response = await domain.process_command(command)
        return response
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        logger.exception("Error creating user")
        raise HTTPException(status_code=500, detail="Internal server error")

Custom Exception Handler​

from fastapi import Request
from fastapi.responses import JSONResponse

class DomainError(Exception):
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.message}
    )

Error Types​

Validation Errors​

from pydantic import ValidationError

try:
    user = UserState(name="", email="invalid")
except ValidationError as e:
    # Handle validation error
    pass

Not Found Errors​

from fastapi import HTTPException

user = await domain.statemgr.fetch('user', user_id)
if not user:
    raise HTTPException(status_code=404, detail="User not found")

Permission Errors​

from fastapi import HTTPException

if not has_permission(user, 'user', 'update'):
    raise HTTPException(status_code=403, detail="Permission denied")

Error Logging​

Structured Logging​

import logging

logger = logging.getLogger(__name__)

try:
    response = await domain.process_command(command)
except Exception as e:
    logger.error("Command processing failed", extra={
        "command": command.name,
        "error": str(e),
        "user_id": request.state.user.id
    })
    raise

Error Recovery​

Retry Logic​

import asyncio

async def process_with_retry(command, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await domain.process_command(command)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

Best Practices​

1. Use Specific Exceptions​

# Good: Specific exception
raise ValueError("Invalid email")

# Bad: Generic exception
raise Exception("Error")

2. Include Context​

# Good: Include context
raise ValueError(f"Invalid email: {email}")

# Bad: No context
raise ValueError("Invalid email")

3. Log Errors​

try:
    response = await domain.process_command(command)
except Exception as e:
    logger.exception("Error processing command")
    raise

4. Return User-Friendly Messages​

try:
    response = await domain.process_command(command)
except ValueError as e:
    # Return user-friendly message
    return {"error": "Please check your input and try again"}

Next Steps​