fastapi-microservice-patterns

FastAPI Microservice Development Patterns

🏗️ SERVICE ARCHITECTURE PATTERNS (From Vibe Coding Backend)

Based on proven patterns from ingestion_orchestration_fastapi_app and ml_inference_fastapi_app.

✅ MANDATORY SERVICE STRUCTURE:

1. Application Lifecycle Management

# ✅ ALWAYS use lifespan context manager for startup/shutdown
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan manager for startup/shutdown."""
    # --- Startup ---
    logger.info("Starting up services...")
    
    # Initialize dependencies (database clients, ML models, etc.)
    app_state.qdrant_client = QdrantClient(host=qdrant_host, port=qdrant_port)
    
    # Perform health checks and capability probing
    _probe_safe_batch_size()  # Example from ML service
    
    logger.info("Startup complete.")
    yield
    
    # --- Shutdown ---
    logger.info("Shutting down services...")
    # Cleanup resources
    if torch.cuda.is_available():
        torch.cuda.empty_cache()
    logger.info("Shutdown complete.")

# ✅ Create app with lifespan
app = FastAPI(title="Your Service Name", lifespan=lifespan)

2. Router Organization Pattern

# ✅ ALWAYS organize endpoints in separate router modules
# Structure: /routers/{domain}.py

# backend/{service}/routers/collections.py
router = APIRouter(prefix="/api/v1/collections", tags=["collections"])

# backend/{service}/main.py
from .routers import search, images, collections, ingest

app.include_router(search.router)
app.include_router(images.router)
app.include_router(collections.router)
app.include_router(ingest.router)

3. Dependencies Module Pattern

# ✅ ALWAYS create centralized dependencies.py
# backend/{service}/dependencies.py

class AppState:
    """Centralized application state container."""
    def __init__(self):
        self.client: SomeClient | None = None
        self.config: Dict[str, Any] = {}

app_state = AppState()

def get_client() -> SomeClient:
    """Dependency function to get initialized client."""
    if app_state.client is None:
        raise RuntimeError("Client has not been initialized.")
    return app_state.client