Posted inDevelopment

Building an Asynchronous REST API with FastAPI and Async SQLAlchemy 2.0: Connection Pool Optimization & Alembic Migrations

Development tutorial - IT technology blog
Development tutorial - IT technology blog

1. Background & Real-World Issues

Our team once faced a tricky production incident: a cleanly written FastAPI application suddenly saw latency spike from 25ms to over 800ms when traffic hit 1,500 req/s. The culprit was a classic one—all Uvicorn workers were blocked by an old synchronous database driver.

FastAPI only achieves peak performance when the entire I/O pipeline—from request handling to the database—runs non-blocking. SQLAlchemy 2.0 thoroughly addresses this challenge with native async support via the asyncpg driver and standardized select() syntax.

However, migrating to async is not as simple as just sprinkling the await keyword around. You will encounter two major pitfalls:

  • Disabled Lazy-Loading: Implicit relationship loading is completely disabled in async mode. Accessing relationships improperly outside the session context immediately raises a MissingGreenlet error.
  • Connection Pool Exhaustion: Misconfigured pool sizes can cause leaked or hanging connections, crashing the API during traffic spikes.

2. Setup & Project Structure

Create a virtual environment with Python 3.10+ and install the required packages:

# Initialize virtual environment
python -m venv .venv
source .venv/bin/activate

# Install FastAPI, asyncpg driver, and Alembic
pip install fastapi==0.110.0 uvicorn[standard]==0.29.0 \
    sqlalchemy==2.0.29 asyncpg==0.29.0 \
    alembic==1.13.1 pydantic-settings==2.2.1

A clean, modular project structure:

app/
├── api/
│   └── v1/
│       └── endpoints/
│           └── items.py
├── core/
│   ├── config.py
│   └── database.py
├── models/
│   └── item.py
├── schemas/
│   └── item.py
└── main.py
alembic/
├── env.py
└── versions/
alembic.ini

3. System Setup

3.1. Configuring Async Engine & Connection Pool

In app/core/database.py, initialize the async engine along with critical pool parameters for production:

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/app_db"

engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_size=20,          # Maintain 20 persistent connections ready in the pool
    max_overflow=10,       # Allow up to 10 additional connections under heavy load
    pool_timeout=30,       # Wait up to 30s before timing out if no connection is available
    pool_recycle=1800,     # Recycle connections after 30 minutes to prevent dropped connections
    pool_pre_ping=True,    # Ping connection to verify it is alive before issuing to a session
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    autoflush=False,
    expire_on_commit=False, # Prevent triggering erroneous lazy-loads after commit
)

class Base(DeclarativeBase):
    pass

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

3.2. Defining Models with SQLAlchemy 2.0 Syntax

In app/models/item.py, use Mapped and mapped_column to take full advantage of type hinting:

from sqlalchemy import String, Integer, DateTime, func
from sqlalchemy.orm import Mapped, mapped_column
from datetime import datetime
from app.core.database import Base

class Item(Base):
    __tablename__ = "items"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
    description: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), 
        server_default=func.now()
    )

3.3. Integrating Async Migrations with Alembic

Initialize the Alembic configuration:

alembic init alembic

By default, alembic/env.py only supports synchronous engines. Update the run_migrations_online function to use the async engine with NullPool:

import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

from app.core.database import Base, DATABASE_URL
import app.models.item  # Register model metadata with Base

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata
config.set_main_option("sqlalchemy.url", DATABASE_URL)

def do_run_migrations(connection: Connection) -> None:
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()

def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())

if context.is_offline_mode():
    pass
else:
    run_migrations_online()

3.4. Building API Endpoints

Define the router in app/api/v1/endpoints/items.py:

from fastapi import APIRouter, Depends, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.core.database import get_db
from app.models.item import Item
from pydantic import BaseModel

router = APIRouter(prefix="/items", tags=["items"])

class ItemCreate(BaseModel):
    title: str
    description: str | None = None

class ItemResponse(ItemCreate):
    id: int
    class Config:
        from_attributes = True

@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(payload: ItemCreate, db: AsyncSession = Depends(get_db)):
    new_item = Item(title=payload.title, description=payload.description)
    db.add(new_item)
    await db.flush()  # Flush SQL statement to DB to populate generated fields (e.g. ID) without closing transaction
    await db.refresh(new_item)
    return new_item

@router.get("/", response_model=list[ItemResponse])
async def list_items(skip: int = 0, limit: int = 20, db: AsyncSession = Depends(get_db)):
    query = select(Item).offset(skip).limit(limit)
    result = await db.execute(query)
    return result.scalars().all()

4. Testing & Production Operations

Generate and apply the initial migration:

alembic revision --autogenerate -m "init_items_table"
alembic upgrade head

Start the Uvicorn server:

uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

To monitor potential connection leaks in production, add an endpoint to inspect the connection pool status directly:

@app.get("/health/db-pool")
async def pool_status():
    pool = engine.pool
    return {
        "pool_size": pool.size(),
        "checked_in": pool.checkedin(),
        "checked_out": pool.checkedout(),
        "overflow": pool.overflow(),
    }

Key operational takeaways to keep in mind:

  • Release sessions immediately: Never call third-party APIs or execute heavy CPU tasks inside an active database session block. Query your data, release the connection, and then proceed with further processing.
  • Calculate pool limits based on worker count: Running Uvicorn with 4 workers (-w 4) creates 4 independent connection pools. With pool_size=20 and max_overflow=10, the app can consume up to 4 x 30 = 120 connections. Ensure PostgreSQL’s max_connections is configured well above this threshold.
  • Enable pool_pre_ping=True: This setting automatically catches and discards stale connections terminated by AWS RDS or network proxies, minimizing ConnectionDoesNotExistError issues.
Share:
Tags: