fastapi-async-microservices

FastAPI Async Microservices Skill Guide

This skill establishes production architecture and coding standards for high-performance, asynchronous REST microservices built with FastAPI, Pydantic v2, and async SQLAlchemy.

---

1. Async Microservice Architecture

[ Client HTTP Request ]
          |
          v
[ Uvicorn ASGI Server (uvloop) ]
          |
          v
[ FastAPI Application Pipeline ]
  |-- Lifespan Handler (DB pool startup / cleanup)
  |-- Global Middleware (CORS, Request-ID, Structured Logging)
  |-- Routers & Dependency Injection (Auth, DB AsyncSession, Rate-limiting)
          |
          +---> [ Pydantic v2 Validation & Deserialization ]
          |
          +---> [ Business Logic & Async SQLAlchemy 2.0 Engine ]
          |                  |
          |                  v (asyncpg connection pool)
          |           [ PostgreSQL 16+ ]
          |
          v
[ Structured JSON Response / RFC 7807 Error ]

---

2. Production Code Standards

A. Environment Configuration (config.py)

from functools import lru_cache
from pydantic import PostgresDsn
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    APP_NAME: str = "OrderService"
    ENV: str = "production"
    DEBUG: bool = False
    PORT: int = 8000

    # Database
    DATABASE_URL: PostgresDsn = "postgresql+asyncpg://user:secret@localhost:5432/orders_db"
    DB_POOL_SIZE: int = 20
    DB_MAX_OVERFLOW: int = 10
    DB_POOL_TIMEOUT: float = 30.0

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=True,
        extra="ignore"
    )


@lru_cache
def get_settings() -> Settings:
    return Settings()

B. Database Session Lifecycle (database.py)

from collections.abc import AsyncGenerator
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from app.config import get_settings

settings = get_settings()

engine = create_async_engine(
    str(settings.DATABASE_URL),
    pool_size=settings.DB_POOL_SIZE,
    max_overflow=settings.DB_MAX_OVERFLOW,
    pool_timeout=settings.DB_POOL_TIMEOUT,
    pool_pre_ping=True,
    echo=settings.DEBUG,
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)


async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    # FastAPI Dependency providing request-scoped async database session
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise