FastAPI Advanced Topics and Best Practices - Interview Questions
1. How do you implement custom middleware in FastAPI?
from fastapi import FastAPI, Request, Response
import time
import json
from typing import Callable
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
# Custom middleware class
class CustomMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next: Callable) -> Response:
# Pre-processing
start_time = time.time()
# Add custom headers
request.state.request_id = f"req_{int(start_time * 1000)}"
# Process request
response = await call_next(request)
# Post-processing
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
response.headers["X-Request-ID"] = request.state.request_id
return response
# Function-based middleware
@app.middleware("http")
async def custom_middleware(request: Request, call_next: Callable):
# Pre-processing
start_time = time.time()
# Log request
print(f"Request: {request.method} {request.url}")
# Process request
response = await call_next(request)
# Post-processing
process_time = time.time() - start_time
print(f"Response: {response.status_code} - {process_time:.3f}s")
return response
# Add middleware to app
app.add_middleware(CustomMiddleware)
# Authentication middleware
class AuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next: Callable) -> Response:
# Skip auth for certain paths
if request.url.path in ["/docs", "/openapi.json", "/health"]:
return await call_next(request)
# Check authentication
auth_header = request.headers.get("Authorization")
if not auth_header:
return Response(
content=json.dumps({"detail": "Authentication required"}),
status_code=401,
media_type="application/json"
)
# Validate token (simplified)
if not auth_header.startswith("Bearer "):
return Response(
content=json.dumps({"detail": "Invalid token format"}),
status_code=401,
media_type="application/json"
)
return await call_next(request)
app.add_middleware(AuthMiddleware)
2. How do you implement custom exception handlers in FastAPI?
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
# Custom exception
class CustomException(Exception):
def __init__(self, message: str, error_code: str):
self.message = message
self.error_code = error_code
# Global exception handler
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
logger.error(f"Custom exception: {exc.message}")
return JSONResponse(
status_code=400,
content={
"error": exc.message,
"error_code": exc.error_code,
"path": request.url.path
}
)
# Validation error handler
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
logger.error(f"Validation error: {exc.errors()}")
return JSONResponse(
status_code=422,
content={
"error": "Validation error",
"details": exc.errors(),
"path": request.url.path
}
)
# HTTP exception handler
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
logger.error(f"HTTP exception: {exc.status_code} - {exc.detail}")
return JSONResponse(
status_code=exc.status_code,
content={
"error": exc.detail,
"status_code": exc.status_code,
"path": request.url.path
}
)
# Generic exception handler
@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)
return JSONResponse(
status_code=500,
content={
"error": "Internal server error",
"path": request.url.path
}
)
# Using custom exceptions
@app.get("/users/{user_id}")
async def get_user(user_id: int):
if user_id < 1:
raise CustomException("Invalid user ID", "INVALID_USER_ID")
if user_id > 1000:
raise HTTPException(status_code=404, detail="User not found")
return {"user_id": user_id, "name": "John Doe"}
3. How do you implement custom response models and serialization?
from fastapi import FastAPI, Response
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Dict, Any
from datetime import datetime
import json
app = FastAPI()
# Custom response model
class CustomResponse(BaseModel):
success: bool
data: Optional[Any] = None
message: Optional[str] = None
timestamp: datetime = Field(default_factory=datetime.utcnow)
request_id: Optional[str] = None
class Config:
json_encoders = {
datetime: lambda v: v.isoformat()
}
# Paginated response model
class PaginatedResponse(BaseModel):
items: List[Any]
total: int
page: int
size: int
pages: int
@validator('pages', pre=True, always=True)
def calculate_pages(cls, v, values):
if 'total' in values and 'size' in values:
return (values['total'] + values['size'] - 1) // values['size']
return v
# Custom response class
class CustomJSONResponse(Response):
def __init__(self, content: Any, status_code: int = 200, **kwargs):
if isinstance(content, dict):
content = CustomResponse(
success=status_code < 400,
data=content,
message=kwargs.pop('message', None)
).dict()
super().__init__(
content=json.dumps(content),
status_code=status_code,
media_type="application/json",
**kwargs
)
# Using custom responses
@app.get("/users", response_model=PaginatedResponse)
async def get_users(page: int = 1, size: int = 10):
# Simulate database query
total_users = 100
users = [{"id": i, "name": f"User {i}"} for i in range((page-1)*size, min(page*size, total_users))]
return PaginatedResponse(
items=users,
total=total_users,
page=page,
size=size
)
@app.get("/user/{user_id}")
async def get_user(user_id: int):
user = {"id": user_id, "name": "John Doe"}
return CustomJSONResponse(
content=user,
message="User retrieved successfully"
)
# Custom serialization
class UserModel(BaseModel):
id: int
name: str
email: str
password: str = Field(..., exclude=True) # Exclude from serialization
class Config:
# Custom JSON serialization
json_encoders = {
datetime: lambda v: v.isoformat()
}
# Schema generation
schema_extra = {
"example": {
"id": 1,
"name": "John Doe",
"email": "john@example.com"
}
}
@app.post("/users", response_model=UserModel)
async def create_user(user: UserModel):
# Password will be excluded from response
return user
4. How do you implement custom path operations and routing?
from fastapi import FastAPI, APIRouter, Request, Response
from fastapi.routing import APIRoute
from typing import Callable, Any
import time
app = FastAPI()
# Custom route class
class TimedRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
start_time = time.time()
response = await original_route_handler(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
return custom_route_handler
# Custom router with middleware
class CustomAPIRouter(APIRouter):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.add_middleware(self.logging_middleware)
async def logging_middleware(self, request: Request, call_next: Callable):
print(f"Router middleware: {request.method} {request.url}")
response = await call_next(request)
print(f"Router response: {response.status_code}")
return response
# Using custom router
users_router = CustomAPIRouter(prefix="/users", tags=["users"])
@users_router.get("/")
async def get_users():
return {"users": ["user1", "user2"]}
@users_router.post("/")
async def create_user():
return {"message": "User created"}
app.include_router(users_router)
# Custom path operation decorator
def rate_limited(requests_per_minute: int = 60):
def decorator(func: Callable) -> Callable:
async def wrapper(*args, **kwargs):
# Rate limiting logic here
return await func(*args, **kwargs)
return wrapper
return decorator
@app.get("/rate-limited")
@rate_limited(requests_per_minute=10)
async def rate_limited_endpoint():
return {"message": "Rate limited endpoint"}
# Custom routing with path parameters
@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
return {"file_path": file_path}
# Custom routing with regex
from fastapi import Path
@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(..., ge=1, le=1000)):
return {"user_id": user_id}
# Custom routing with query parameters
from fastapi import Query
@app.get("/search")
async def search_items(
q: str = Query(..., min_length=1, max_length=100),
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100)
):
return {"q": q, "skip": skip, "limit": limit}
5. How do you implement custom dependency providers?
from fastapi import FastAPI, Depends, HTTPException
from typing import Optional, Callable, Any
from functools import wraps
import asyncio
app = FastAPI()
# Custom dependency provider
class DependencyProvider:
def __init__(self):
self._dependencies = {}
def register(self, name: str, provider: Callable):
self._dependencies[name] = provider
def get(self, name: str) -> Any:
if name not in self._dependencies:
raise ValueError(f"Dependency '{name}' not found")
return self._dependencies[name]()
# Global dependency provider
dependency_provider = DependencyProvider()
# Register dependencies
dependency_provider.register("database", lambda: {"type": "postgresql"})
dependency_provider.register("cache", lambda: {"type": "redis"})
# Custom dependency function
def get_dependency(name: str):
def dependency():
return dependency_provider.get(name)
return dependency
# Using custom dependencies
@app.get("/config")
async def get_config(
db_config: dict = Depends(get_dependency("database")),
cache_config: dict = Depends(get_dependency("cache"))
):
return {"database": db_config, "cache": cache_config}
# Async dependency provider
class AsyncDependencyProvider:
def __init__(self):
self._dependencies = {}
def register(self, name: str, provider: Callable):
self._dependencies[name] = provider
async def get(self, name: str) -> Any:
if name not in self._dependencies:
raise ValueError(f"Dependency '{name}' not found")
provider = self._dependencies[name]
if asyncio.iscoroutinefunction(provider):
return await provider()
return provider()
async_dependency_provider = AsyncDependencyProvider()
async def get_async_database():
await asyncio.sleep(0.1) # Simulate async operation
return {"type": "async_postgresql"}
async_dependency_provider.register("async_database", get_async_database)
def get_async_dependency(name: str):
async def dependency():
return await async_dependency_provider.get(name)
return dependency
@app.get("/async-config")
async def get_async_config(
db_config: dict = Depends(get_async_dependency("async_database"))
):
return {"database": db_config}
# Dependency with parameters
def create_dependency_with_params(param_name: str, default_value: Any = None):
def dependency(param: str = Query(default=default_value, alias=param_name)):
return {param_name: param}
return dependency
@app.get("/parametrized")
async def parametrized_endpoint(
user_config: dict = Depends(create_dependency_with_params("user_id")),
session_config: dict = Depends(create_dependency_with_params("session_id"))
):
return {"user": user_config, "session": session_config}
6. How do you implement custom background tasks and job queues?
from fastapi import FastAPI, BackgroundTasks
import asyncio
import threading
from queue import Queue
from typing import Dict, Any
import time
app = FastAPI()
# Custom background task manager
class BackgroundTaskManager:
def __init__(self):
self.tasks: Dict[str, asyncio.Task] = {}
self.task_queue = Queue()
self.running = True
self.worker_thread = threading.Thread(target=self._worker, daemon=True)
self.worker_thread.start()
def _worker(self):
while self.running:
try:
task = self.task_queue.get(timeout=1)
if task:
task()
self.task_queue.task_done()
except Exception as e:
print(f"Task error: {e}")
def add_task(self, task_id: str, task_func: Callable):
self.task_queue.put(task_func)
self.tasks[task_id] = None # Placeholder
def stop(self):
self.running = False
self.worker_thread.join()
task_manager = BackgroundTaskManager()
# Custom background task decorator
def background_task(task_name: str):
def decorator(func: Callable):
@wraps(func)
async def wrapper(*args, **kwargs):
task_id = f"{task_name}_{int(time.time())}"
task_manager.add_task(task_id, lambda: func(*args, **kwargs))
return {"task_id": task_id, "status": "queued"}
return wrapper
return decorator
@app.post("/process-data")
@background_task("data_processing")
async def process_data(data: dict):
# This will run in background
time.sleep(5) # Simulate processing
print(f"Processed data: {data}")
return {"status": "processed"}
# Async background tasks
class AsyncTaskManager:
def __init__(self):
self.tasks: Dict[str, asyncio.Task] = {}
async def add_task(self, task_id: str, task_func: Callable):
task = asyncio.create_task(task_func())
self.tasks[task_id] = task
return task_id
async def get_task_status(self, task_id: str):
if task_id not in self.tasks:
return {"status": "not_found"}
task = self.tasks[task_id]
if task.done():
try:
result = task.result()
return {"status": "completed", "result": result}
except Exception as e:
return {"status": "failed", "error": str(e)}
else:
return {"status": "running"}
async_task_manager = AsyncTaskManager()
async def long_running_task(data: dict):
await asyncio.sleep(10) # Simulate long task
return {"processed": data}
@app.post("/async-process")
async def async_process_data(data: dict):
task_id = await async_task_manager.add_task(
f"async_process_{int(time.time())}",
lambda: long_running_task(data)
)
return {"task_id": task_id}
@app.get("/task-status/{task_id}")
async def get_task_status(task_id: str):
return await async_task_manager.get_task_status(task_id)
# Custom job queue with priorities
from heapq import heappush, heappop
from dataclasses import dataclass
from typing import Any
@dataclass
class Job:
priority: int
task_id: str
func: Callable
args: tuple
kwargs: dict
def __lt__(self, other):
return self.priority < other.priority
class PriorityJobQueue:
def __init__(self):
self.queue = []
self.running = True
self.worker_thread = threading.Thread(target=self._worker, daemon=True)
self.worker_thread.start()
def _worker(self):
while self.running:
if self.queue:
job = heappop(self.queue)
try:
job.func(*job.args, **job.kwargs)
except Exception as e:
print(f"Job error: {e}")
else:
time.sleep(0.1)
def add_job(self, priority: int, task_id: str, func: Callable, *args, **kwargs):
job = Job(priority, task_id, func, args, kwargs)
heappush(self.queue, job)
return task_id
priority_queue = PriorityJobQueue()
@app.post("/priority-task")
async def add_priority_task(priority: int, data: dict):
def process_priority_task(data: dict):
time.sleep(2)
print(f"Processed priority task: {data}")
task_id = priority_queue.add_job(
priority,
f"priority_task_{int(time.time())}",
process_priority_task,
data
)
return {"task_id": task_id}
7. How do you implement custom OpenAPI documentation?
from fastapi import FastAPI, APIRouter
from fastapi.openapi.utils import get_openapi
from typing import Dict, Any
app = FastAPI(
title="Custom API",
description="A custom API with enhanced documentation",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc"
)
# Custom OpenAPI schema
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Custom API",
version="1.0.0",
description="Enhanced API documentation",
routes=app.routes,
)
# Add custom components
openapi_schema["components"]["schemas"]["CustomResponse"] = {
"type": "object",
"properties": {
"success": {"type": "boolean"},
"data": {"type": "object"},
"message": {"type": "string"},
"timestamp": {"type": "string", "format": "date-time"}
}
}
# Add custom security schemes
openapi_schema["components"]["securitySchemes"] = {
"CustomAuth": {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT",
"description": "Custom JWT authentication"
}
}
# Add custom tags
openapi_schema["tags"] = [
{
"name": "users",
"description": "User management operations",
"externalDocs": {
"description": "User API Documentation",
"url": "https://example.com/docs/users"
}
},
{
"name": "admin",
"description": "Administrative operations"
}
]
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
# Custom documentation examples
from pydantic import BaseModel, Field
class UserCreate(BaseModel):
name: str = Field(..., example="John Doe", description="User's full name")
email: str = Field(..., example="john@example.com", description="User's email address")
age: int = Field(..., ge=0, le=120, example=30, description="User's age")
class Config:
schema_extra = {
"example": {
"name": "John Doe",
"email": "john@example.com",
"age": 30
}
}
@app.post("/users",
response_model=UserCreate,
tags=["users"],
summary="Create a new user",
description="Create a new user with the provided information",
response_description="The created user",
responses={
200: {
"description": "User created successfully",
"content": {
"application/json": {
"example": {
"name": "John Doe",
"email": "john@example.com",
"age": 30
}
}
}
},
422: {
"description": "Validation error",
"content": {
"application/json": {
"example": {
"detail": [
{
"loc": ["body", "email"],
"msg": "invalid email address",
"type": "value_error.email"
}
]
}
}
}
}
}
)
async def create_user(user: UserCreate):
"""
Create a new user with the following information:
- **name**: User's full name
- **email**: User's email address (must be valid)
- **age**: User's age (must be between 0 and 120)
Returns the created user information.
"""
return user
# Custom router with documentation
users_router = APIRouter(
prefix="/users",
tags=["users"],
responses={404: {"description": "User not found"}}
)
@users_router.get("/{user_id}",
summary="Get user by ID",
description="Retrieve a user by their unique identifier",
response_description="User information"
)
async def get_user(user_id: int):
return {"user_id": user_id, "name": "John Doe"}
app.include_router(users_router)
# Custom documentation endpoint
@app.get("/api-docs")
async def get_api_docs():
return {
"title": "Custom API Documentation",
"version": "1.0.0",
"endpoints": [
{
"path": "/users",
"method": "POST",
"description": "Create a new user"
},
{
"path": "/users/{user_id}",
"method": "GET",
"description": "Get user by ID"
}
]
}
8. How do you implement custom request/response processing?
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponse
from typing import Callable, Any
import json
import gzip
import base64
app = FastAPI()
# Custom request processing
class CustomRequestProcessor:
def __init__(self):
self.processors = []
def add_processor(self, processor: Callable):
self.processors.append(processor)
async def process(self, request: Request):
for processor in self.processors:
request = await processor(request)
return request
request_processor = CustomRequestProcessor()
# Add request processors
async def add_request_id(request: Request):
request.state.request_id = f"req_{int(time.time() * 1000)}"
return request
async def log_request(request: Request):
print(f"Processing request: {request.method} {request.url}")
return request
request_processor.add_processor(add_request_id)
request_processor.add_processor(log_request)
# Custom response processing
class CustomResponseProcessor:
def __init__(self):
self.processors = []
def add_processor(self, processor: Callable):
self.processors.append(processor)
async def process(self, response: Response, request: Request):
for processor in self.processors:
response = await processor(response, request)
return response
response_processor = CustomResponseProcessor()
# Add response processors
async def add_response_headers(response: Response, request: Request):
response.headers["X-Request-ID"] = getattr(request.state, "request_id", "unknown")
response.headers["X-Processed-By"] = "CustomProcessor"
return response
async def compress_response(response: Response, request: Request):
if "gzip" in request.headers.get("accept-encoding", ""):
content = response.body
compressed = gzip.compress(content)
response.body = compressed
response.headers["content-encoding"] = "gzip"
response.headers["content-length"] = str(len(compressed))
return response
response_processor.add_processor(add_response_headers)
response_processor.add_processor(compress_response)
# Middleware to apply processors
@app.middleware("http")
async def apply_processors(request: Request, call_next: Callable):
# Process request
request = await request_processor.process(request)
# Get response
response = await call_next(request)
# Process response
response = await response_processor.process(response, request)
return response
# Custom request body processing
class CustomRequestBodyProcessor:
def __init__(self):
self.processors = {}
def register_processor(self, content_type: str, processor: Callable):
self.processors[content_type] = processor
async def process_body(self, request: Request):
content_type = request.headers.get("content-type", "")
if content_type in self.processors:
body = await request.body()
processed_body = await self.processors[content_type](body)
request._body = processed_body
return request
body_processor = CustomRequestBodyProcessor()
# Register body processors
async def process_json_body(body: bytes):
data = json.loads(body)
# Add timestamp to JSON data
data["timestamp"] = time.time()
return json.dumps(data).encode()
async def process_base64_body(body: bytes):
# Decode base64 body
decoded = base64.b64decode(body)
return decoded
body_processor.register_processor("application/json", process_json_body)
body_processor.register_processor("application/base64", process_base64_body)
# Custom response formatting
class CustomResponseFormatter:
def __init__(self):
self.formatters = {}
def register_formatter(self, content_type: str, formatter: Callable):
self.formatters[content_type] = formatter
def format_response(self, content: Any, content_type: str = "application/json"):
if content_type in self.formatters:
return self.formatters[content_type](content)
return content
response_formatter = CustomResponseFormatter()
# Register response formatters
def format_json_response(content: Any):
return {
"success": True,
"data": content,
"timestamp": time.time()
}
def format_xml_response(content: Any):
# Simple XML formatting
return f"<response><data>{content}</data></response>"
response_formatter.register_formatter("application/json", format_json_response)
response_formatter.register_formatter("application/xml", format_xml_response)
# Using custom processors
@app.post("/custom-process")
async def custom_process_endpoint(request: Request):
# Process request body
request = await body_processor.process_body(request)
# Get request data
body = await request.json()
# Format response
formatted_response = response_formatter.format_response(body)
return JSONResponse(content=formatted_response)
9. How do you implement custom validation and business logic?
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, validator, root_validator
from typing import List, Optional, Dict, Any
import re
app = FastAPI()
# Custom validation decorator
def validate_business_rules(*rules):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# Apply business rules
for rule in rules:
result = await rule(*args, **kwargs)
if not result["valid"]:
raise HTTPException(
status_code=400,
detail=result["message"]
)
return await func(*args, **kwargs)
return wrapper
return decorator
# Business rule functions
async def validate_user_age(data):
if "age" in data and data["age"] < 18:
return {"valid": False, "message": "User must be at least 18 years old"}
return {"valid": True}
async def validate_email_domain(data):
if "email" in data and not data["email"].endswith("@company.com"):
return {"valid": False, "message": "Email must be from company domain"}
return {"valid": True}
# Custom model with business validation
class UserWithBusinessRules(BaseModel):
name: str
email: str
age: int
department: str
@validator('name')
def validate_name(cls, v):
if len(v.strip()) < 2:
raise ValueError('Name must be at least 2 characters')
if not re.match(r'^[a-zA-Z\s]+$', v):
raise ValueError('Name can only contain letters and spaces')
return v.title()
@validator('email')
def validate_email(cls, v):
if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$', v):
raise ValueError('Invalid email format')
return v.lower()
@validator('age')
def validate_age(cls, v):
if v < 18 or v > 65:
raise ValueError('Age must be between 18 and 65')
return v
@root_validator
def validate_business_rules(cls, values):
# Cross-field validation
if 'department' in values and 'age' in values:
if values['department'] == 'management' and values['age'] < 25:
raise ValueError('Management positions require minimum age of 25')
return values
# Custom validation service
class ValidationService:
def __init__(self):
self.validators = {}
def register_validator(self, name: str, validator_func: Callable):
self.validators[name] = validator_func
async def validate(self, name: str, data: Any) -> Dict[str, Any]:
if name not in self.validators:
return {"valid": True, "message": "No validator found"}
return await self.validators[name](data)
validation_service = ValidationService()
# Register validators
async def validate_user_data(data: Dict[str, Any]):
errors = []
if "email" in data and "@" not in data["email"]:
errors.append("Invalid email format")
if "age" in data and (data["age"] < 0 or data["age"] > 150):
errors.append("Invalid age")
return {
"valid": len(errors) == 0,
"message": "; ".join(errors) if errors else "Valid"
}
validation_service.register_validator("user_data", validate_user_data)
# Using custom validation
@app.post("/users")
@validate_business_rules(validate_user_age, validate_email_domain)
async def create_user(user: UserWithBusinessRules):
return {"message": "User created", "user": user.dict()}
@app.post("/validate")
async def validate_data(data: Dict[str, Any]):
result = await validation_service.validate("user_data", data)
return result
# Custom business logic service
class BusinessLogicService:
def __init__(self):
self.rules = {}
def add_rule(self, name: str, rule_func: Callable):
self.rules[name] = rule_func
async def apply_rules(self, data: Dict[str, Any]) -> Dict[str, Any]:
for rule_name, rule_func in self.rules.items():
data = await rule_func(data)
return data
business_service = BusinessLogicService()
# Business rule functions
async def apply_discount_rule(data: Dict[str, Any]) -> Dict[str, Any]:
if "total_amount" in data and data["total_amount"] > 100:
data["discount"] = data["total_amount"] * 0.1
data["final_amount"] = data["total_amount"] - data["discount"]
return data
async def apply_tax_rule(data: Dict[str, Any]) -> Dict[str, Any]:
if "final_amount" in data:
data["tax"] = data["final_amount"] * 0.08
data["total_with_tax"] = data["final_amount"] + data["tax"]
return data
business_service.add_rule("discount", apply_discount_rule)
business_service.add_rule("tax", apply_tax_rule)
@app.post("/calculate-total")
async def calculate_total(order: Dict[str, Any]):
result = await business_service.apply_rules(order)
return result
10. How do you implement custom error handling and recovery?
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from typing import Dict, Any, Optional, Callable
import logging
import traceback
import asyncio
app = FastAPI()
# Custom error handler
class CustomErrorHandler:
def __init__(self):
self.error_handlers = {}
self.recovery_strategies = {}
self.logger = logging.getLogger(__name__)
def register_error_handler(self, error_type: type, handler: Callable):
self.error_handlers[error_type] = handler
def register_recovery_strategy(self, error_type: type, strategy: Callable):
self.recovery_strategies[error_type] = strategy
async def handle_error(self, error: Exception, request: Request) -> JSONResponse:
error_type = type(error)
# Log error
self.logger.error(f"Error occurred: {error}", exc_info=True)
# Try to handle with custom handler
if error_type in self.error_handlers:
return await self.error_handlers[error_type](error, request)
# Try recovery strategy
if error_type in self.recovery_strategies:
try:
result = await self.recovery_strategies[error_type](error, request)
if result:
return result
except Exception as recovery_error:
self.logger.error(f"Recovery failed: {recovery_error}")
# Default error response
return JSONResponse(
status_code=500,
content={
"error": "Internal server error",
"type": error_type.__name__,
"message": str(error)
}
)
error_handler = CustomErrorHandler()
# Custom error types
class ValidationError(Exception):
def __init__(self, message: str, field: str):
self.message = message
self.field = field
class BusinessRuleError(Exception):
def __init__(self, message: str, rule: str):
self.message = message
self.rule = rule
class ExternalServiceError(Exception):
def __init__(self, service: str, message: str):
self.service = service
self.message = message
# Register error handlers
async def handle_validation_error(error: ValidationError, request: Request):
return JSONResponse(
status_code=400,
content={
"error": "Validation error",
"field": error.field,
"message": error.message
}
)
async def handle_business_rule_error(error: BusinessRuleError, request: Request):
return JSONResponse(
status_code=422,
content={
"error": "Business rule violation",
"rule": error.rule,
"message": error.message
}
)
async def handle_external_service_error(error: ExternalServiceError, request: Request):
return JSONResponse(
status_code=503,
content={
"error": "External service unavailable",
"service": error.service,
"message": error.message
}
)
error_handler.register_error_handler(ValidationError, handle_validation_error)
error_handler.register_error_handler(BusinessRuleError, handle_business_rule_error)
error_handler.register_error_handler(ExternalServiceError, handle_external_service_error)
# Recovery strategies
async def retry_strategy(error: Exception, request: Request, max_retries: int = 3):
for attempt in range(max_retries):
try:
# Retry the operation
await asyncio.sleep(2 ** attempt) # Exponential backoff
# Re-execute the original request logic
return await handle_request(request)
except Exception as retry_error:
if attempt == max_retries - 1:
raise retry_error
async def fallback_strategy(error: Exception, request: Request):
# Return fallback response
return JSONResponse(
status_code=200,
content={
"message": "Fallback response",
"original_error": str(error)
}
)
error_handler.register_recovery_strategy(ExternalServiceError, retry_strategy)
error_handler.register_recovery_strategy(ValidationError, fallback_strategy)
# Middleware for error handling
@app.middleware("http")
async def error_handling_middleware(request: Request, call_next: Callable):
try:
response = await call_next(request)
return response
except Exception as error:
return await error_handler.handle_error(error, request)
# Using custom errors
@app.post("/users")
async def create_user(user_data: Dict[str, Any]):
if "email" not in user_data:
raise ValidationError("Email is required", "email")
if user_data.get("age", 0) < 18:
raise BusinessRuleError("User must be at least 18 years old", "age_requirement")
# Simulate external service call
try:
# External service call
pass
except Exception as e:
raise ExternalServiceError("user_service", "Failed to create user")
return {"message": "User created"}
# Circuit breaker pattern
class CircuitBreaker:
def __init__(self, failure_threshold: int = 5, timeout: int = 60):
self.failure_threshold = failure_threshold
self.timeout = timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
async def call(self, func: Callable, *args, **kwargs):
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.timeout:
self.state = "HALF_OPEN"
else:
raise Exception("Circuit breaker is OPEN")
try:
result = await func(*args, **kwargs)
if self.state == "HALF_OPEN":
self.state = "CLOSED"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
raise e
circuit_breaker = CircuitBreaker()
@app.get("/external-data")
async def get_external_data():
async def external_call():
# Simulate external API call
if random.random() < 0.3: # 30% failure rate
raise ExternalServiceError("external_api", "Service unavailable")
return {"data": "external data"}
return await circuit_breaker.call(external_call)
Interview angle
- “Middleware or a dependency?” - middleware for cross-cutting concerns on every request (correlation IDs, timing, CORS); a dependency when it needs the parsed request, applies to a subset of routes, or should produce a typed value the handler consumes.
- “How do you customise error responses?” - exception handlers registered on the app, mapping domain exceptions to HTTP responses. That keeps HTTP status decisions out of business logic while still producing consistent error bodies.
- “How do you stream a response?” -
StreamingResponse with an async generator, which is how SSE and LLM token streaming work. Watch for proxy buffering, which silently defeats it. See ../../12_protocols/sse/01_server_sent_events.md.
- “When would you use
BackgroundTasks?” - only for work you can afford to lose, because it runs in-process and dies with it. Anything the business needs to have happened belongs on a durable queue.