Connection pooling in FastAPI

Introduction

Connection pooling helps your FastAPI app reuse database connections. This makes your app faster and uses fewer resources.

When your app talks to a database many times per second.

When you want to avoid the delay of opening a new database connection each time.

When you want to limit how many database connections your app uses at once.

When you want your app to handle many users smoothly without slowing down.

Syntax

FastAPI

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/dbname",
    pool_size=5,  # Number of connections to keep in the pool
    max_overflow=10,  # Extra connections allowed beyond pool_size
)

async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)

Use create_async_engine to create a database engine with connection pooling.

pool_size controls how many connections stay open ready to use.

Examples

This sets a pool of 10 connections and allows 20 extra temporary connections if needed.

FastAPI

engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/dbname",
    pool_size=10,
    max_overflow=20
)

This creates a session factory that uses the connection pool to get connections.

FastAPI

async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)

Sample Program

This FastAPI app uses connection pooling to talk to a SQLite database asynchronously. The get_session function provides a session from the pool. The endpoint runs a simple query and returns the result.

FastAPI

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text

app = FastAPI()

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

engine = create_async_engine(
    DATABASE_URL
)

async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)

async def get_session() -> AsyncSession:
    async with async_session() as session:
        yield session

@app.get("/items")
async def read_items(session: AsyncSession = Depends(get_session)):
    result = await session.execute(text("SELECT 1 as number"))
    number = result.scalar_one()
    return {"number": number}

Important Notes

Connection pools keep connections open so your app can reuse them quickly.

Adjust pool_size and max_overflow based on your app's load and database limits.

Always close sessions properly to return connections to the pool.

Summary

Connection pooling speeds up database access by reusing connections.

FastAPI with SQLAlchemy supports connection pooling easily.

Configure pool size to balance performance and resource use.