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 asyncpgIf you plan to use SQLAlchemy later, also install:
pip install sqlalchemy[asyncio] psycopg2-binary2. Minimal FastAPI app using asyncpg
Create app/main.py:
from fastapi import FastAPI, HTTPExceptionimport asyncpgimport 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 --reloadYou 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
Recordobjects 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] asyncpg2. Configure the async engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSessionfrom 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 session5. 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 backSQLAlchemy 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 Requestfrom fastapi.responses import JSONResponseimport 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 csvfrom 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_connectionsshould be at leastpool.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-slimWORKDIR /appCOPY requirements.txt .RUN pip install -r requirements.txtCOPY . .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: 30Key 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()orAsyncSession.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!
Working on something similar?
If you're building backend or AI systems and want a second set of senior eyes, let's talk.
