Supercharge Your API: FastAPI asyncpg Integration for High‑Performance PostgreSQL Access

Introduction

If you’re building a modern Python web service, FastAPI asyncpg is the duo you want to master. FastAPI gives you an ultra‑fast, type‑safe HTTP layer, while asyncpg delivers low‑overhead, fully asynchronous PostgreSQL access. Together they let you serve thousands of requests per second with minimal latency. In this post we’ll walk through setting up asyncpg inside a FastAPI project, manage connection pools correctly, hook asyncpg into SQLAlchemy 2.0’s async engine, handle transactions robustly, and finally squeeze out every last drop of performance for production deployments.

Note: This guide assumes Python 3.9+, FastAPI 0.100+, asyncpg 0.28+, and SQLAlchemy 2.0. If you’re still on the classic SQLAlchemy ORM, check out our article on Fix AsyncSession Errors in FastAPI (SQLAlchemy 2.0 Guide).

---

Setting up asyncpg in a FastAPI project

1. Install the dependencies

pip install fastapi uvicorn asyncpg

If you plan to use SQLAlchemy later, also install:

pip install sqlalchemy[asyncio] psycopg2-binary

2. Minimal FastAPI app using asyncpg

Create app/main.py:

from fastapi import FastAPI, HTTPException
import asyncpg
import os

app = FastAPI()

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql://postgres:password@localhost:5432/mydb"
)

# We'll fill the pool in the startup event (see next section)
db_pool: asyncpg.Pool | None = None


@app.on_event("startup")
async def startup():
    global db_pool
    db_pool = await asyncpg.create_pool(DATABASE_URL, min_size=5, max_size=20)
    # Test connection
    async with db_pool.acquire() as conn:
        await conn.execute("SELECT 1")


@app.on_event("shutdown")
async def shutdown():
    global db_pool
    if db_pool:
        await db_pool.close()


@app.get("/users/{user_id}")
async def read_user(user_id: int):
    async with db_pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, username, email FROM users WHERE id = $1", user_id
        )
        if not row:
            raise HTTPException(status_code=404, detail="User not found")
        return dict(row)

Run it:

uvicorn app.main:app --reload

You now have a FastAPI asyncpg endpoint that pulls a user record without ever blocking the event loop.

Why asyncpg?

• Native async/await – built on libpq’s asynchronous API, no thread‑pool shim.
• Prepared statements – automatic caching of query plans.
• Fast serialization – returns Record objects that map cleanly to dicts.

---

Connection pooling and lifecycle management with FastAPI

Creating a new PostgreSQL connection per request is a serious performance killer. asyncpg’s built‑in connection pool solves this, but you need to tie its lifecycle to FastAPI’s startup/shutdown events.

Pool configuration best practices

Parameter	Typical value	Reason
min_size	5	Keep a few warm connections ready for bursts.
max_size	20‑50 (depends on DB limits)	Prevent exhausting DB resources under load.
max_queries	5000	Re‑prepare statements after this many queries.
max_inactive_connection_lifetime	300	Recycle idle connections to avoid stale sockets.

db_pool = await asyncpg.create_pool(
    DATABASE_URL,
    min_size=5,
    max_size=30,
    max_queries=5000,
    max_inactive_connection_lifetime=300,
    command_timeout=60,
)

Using a Dependency for Per‑request Access

FastAPI’s dependency system lets you inject a connection automatically:

from fastapi import Depends

async def get_connection() -> asyncpg.Connection:
    async with db_pool.acquire() as conn:
        yield conn

@app.get("/posts")
async def list_posts(conn: asyncpg.Connection = Depends(get_connection)):
    rows = await conn.fetch("SELECT id, title FROM posts ORDER BY created_at DESC LIMIT 20")
    return [dict(r) for r in rows]

The async with inside the dependency guarantees the connection is released back to the pool even if the endpoint raises an exception.

Detecting leaks

If you notice “QueuePool limit reached” errors, you probably have connections not returning to the pool. The article Fixing “QueuePool limit reached”: Debugging DB Connection Leaks in FastAPI explains how to instrument asyncpg’s pool statistics to spot leaks early.

---

Using asyncpg with SQLAlchemy 2.0 async engine

SQLAlchemy 2.0 introduced a first‑class async API that can sit on top of asyncpg. This gives you the expressive ORM while retaining raw‑SQL performance where needed.

1. Install the async driver

SQLAlchemy will automatically use asyncpg when you specify the postgresql+asyncpg dialect.

pip install sqlalchemy[asyncio] asyncpg

2. Configure the async engine

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

DATABASE_URL_ASYNC = (
    "postgresql+asyncpg://postgres:password@localhost:5432/mydb"
)

engine = create_async_engine(
    DATABASE_URL_ASYNC,
    pool_size=20,
    max_overflow=10,
    future=True,  # enables 2.0 style usage
)

AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)
Base = declarative_base()

3. Define a model

from sqlalchemy import Column, Integer, String

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(120), unique=True, nullable=False)

4. Dependency that yields an async session

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

5. Mixing raw asyncpg and ORM

You can still run raw asyncpg queries inside the same request:

@app.get("/users/{user_id}/stats")
async def user_stats(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    conn: asyncpg.Connection = Depends(get_connection),  # from previous section
):
    # ORM part
    result = await db.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")

    # Raw asyncpg part – heavy analytics query
    stats = await conn.fetchrow(
        """
        SELECT count(*) AS post_count,
               avg(like_count) AS avg_likes
        FROM posts
        WHERE author_id = $1
        """,
        user_id,
    )
    return {
        "user": {"id": user.id, "username": user.username},
        "stats": dict(stats),
    }

This hybrid approach keeps the codebase flexible: use the ORM for everyday CRUD, and asyncpg for high‑performance analytics.

---

Transaction handling and error management

When you work with asyncpg directly, you must manage transactions explicitly. SQLAlchemy’s AsyncSession does it for you, but you still need to be aware of proper patterns.

Using asyncpg transaction objects

@app.post("/transfer")
async def transfer_funds(
    from_user: int,
    to_user: int,
    amount: float,
    conn: asyncpg.Connection = Depends(get_connection),
):
    async with conn.transaction():
        # Debit
        await conn.execute(
            """
            UPDATE accounts
            SET balance = balance - $1
            WHERE user_id = $2 AND balance >= $1
            """,
            amount,
            from_user,
        )
        # Credit
        await conn.execute(
            """
            UPDATE accounts
            SET balance = balance + $1
            WHERE user_id = $2
            """,
            amount,
            to_user,
        )
    return {"status": "success"}

If any statement raises an exception, the async with conn.transaction() block rolls back automatically.

Nesting and savepoints

PostgreSQL supports savepoints, which asyncpg exposes via transaction() with nested=True:

async with conn.transaction():
    await conn.execute("INSERT INTO orders ...")
    async with conn.transaction(nested=True):
        # This can be rolled back independently
        await conn.execute("INSERT INTO order_items ...")
        # raise some error -> only inner transaction rolls back

SQLAlchemy async transaction pattern

@app.post("/orders")
async def create_order(
    order_data: OrderCreate,
    db: AsyncSession = Depends(get_db),
):
    async with db.begin():
        new_order = Order(**order_data.dict())
        db.add(new_order)
        # Flush to get the primary key
        await db.flush()
        # Add line items
        for item in order_data.items:
            db.add(OrderItem(order_id=new_order.id, **item.dict()))
    # Commit happens automatically when exiting the `begin` block
    return {"order_id": new_order.id}

If any exception bubbles out, the transaction is rolled back.

Centralized error handling

FastAPI lets you register exception handlers. For DB‑related errors you can translate asyncpg.PostgresError into clean HTTP responses:

from fastapi import Request
from fastapi.responses import JSONResponse
import asyncpg

@app.exception_handler(asyncpg.PostgresError)
async def pg_error_handler(request: Request, exc: asyncpg.PostgresError):
    return JSONResponse(
        status_code=400,
        content={"detail": f"Database error: {exc.message}"}
    )

Now any uncaught asyncpg error becomes a 400 response rather than a 500.

---

Performance tuning and best practices for production deployments

1. Use prepared statements wisely

asyncpg automatically prepares statements after the first execution. For hot paths you can pre‑prepare them during startup:

async def preload_statements():
    async with db_pool.acquire() as conn:
        await conn.prepare(
            "SELECT id, username FROM users WHERE id = $1"
        )
        await conn.prepare(
            "INSERT INTO logs (level, message) VALUES ($1, $2)"
        )

Call await preload_statements() inside the startup event.

2. Leverage COPY for bulk loads

If you need to ingest large CSVs, asyncpg’s copy_records_to_table is dramatically faster than row‑by‑row inserts.

import csv
from pathlib import Path

async def bulk_insert_users(csv_path: Path):
    async with db_pool.acquire() as conn:
        with csv_path.open() as f:
            reader = csv.DictReader(f)
            await conn.copy_records_to_table(
                "users",
                records=[(r["username"], r["email"]) for r in reader],
                columns=("username", "email")
            )

3. Monitor pool metrics

Add a simple endpoint to expose pool statistics (useful with Prometheus):

@app.get("/db/pool-stats")
async def pool_stats():
    return {
        "size": db_pool.get_size(),
        "used": db_pool.get_used_size(),
        "free": db_pool.get_free_size(),
        "max_size": db_pool.max_size,
    }

4. Tune PostgreSQL parameters

• max_connections should be at least pool.max_size + some headroom.
• shared_buffers = 25‑30 % of RAM.
• effective_cache_size = 50‑75 % of RAM.

5. Deploy with an ASGI server that respects async I/O

Use uvicorn with the --workers flag only when you have CPU‑bound work; for pure async I/O a single worker is often enough. Example Dockerfile snippet:

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Scale horizontally with a container orchestrator (Kubernetes, Docker Swarm). Each replica gets its own pool, so keep max_connections in mind.

6. Use connection‑level statement timeouts

await conn.execute(
    "SET statement_timeout = 5000"  # 5 seconds
)

This prevents runaway queries from hogging a connection.

7. Periodic health checks

Kubernetes liveness probes can run a lightweight SELECT 1 against the pool:

livenessProbe:
  exec:
    command: ["python", "-c", "import asyncpg, asyncio, os; asyncio.run(asyncpg.connect(os.getenv('DATABASE_URL')).execute('SELECT 1'))"]
  initialDelaySeconds: 10
  periodSeconds: 30

---

Key Takeaways

• FastAPI asyncpg gives you a truly non‑blocking stack for PostgreSQL.
• Create a single asyncpg pool at startup, release connections via async with, and expose it through FastAPI dependencies.
• When you need an ORM, plug asyncpg into SQLAlchemy 2.0’s async engine (postgresql+asyncpg). You can mix raw asyncpg queries with ORM models seamlessly.
• Manage transactions explicitly with conn.transaction() or AsyncSession.begin(). Use nested transactions for fine‑grained rollbacks.
• Tune the pool (min_size, max_size, max_queries), pre‑prepare hot statements, and monitor pool metrics to avoid leaks.
• Align PostgreSQL server settings (max_connections, shared_buffers) with your pool size, and use container orchestration health checks for resiliency.

By following these patterns, your FastAPI services will handle PostgreSQL traffic at lightning speed while staying robust, maintainable, and ready for production scale. Happy coding!