FastAPI Async Database Patterns
Implement async database operations with SQLAlchemy 2.0 for high-concurrency FastAPI applications.
Overview
FastAPI's async capabilities shine brightest when combined with async database drivers and proper connection pooling. Traditional synchronous database drivers block the event loop during queries, negating FastAPI's concurrency benefits. Using async drivers like asyncpg for PostgreSQL or aiomysql for MySQL, combined with SQLAlchemy 2.0's async support, enables truly non-blocking database operations. The async session pattern involves creating a database dependency that provides an AsyncSession scoped to each request. This session should use a connection pool sized appropriately for the expected concurrency level—too small causes connection starvation, too large wastes resources. SQLAlchemy's async engine uses NullPool by default for individual sessions but can be configured with QueuePool for connection reuse. Transaction management in async contexts requires careful handling. Using async with for session.begin() ensures transactions are committed on success or rolled back on exception. For read-heavy workloads, choosing appropriate isolation levels can significantly impact performance. SERIALIZABLE provides strongest guarantees but lowest concurrency, while READ COMMITTED is suitable for most applications. Bulk operations benefit greatly from async patterns. Using session.execute() with Core insert statements for bulk inserts, or batch update statements, reduces round-trips dramatically compared to individual ORM operations. For data migrations or initial data loading, these bulk operations can be orders of magnitude faster than row-by-row processing.
Code Example
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base
from sqlalchemy import Column, Integer, String, DateTime, select
from contextlib import asynccontextmanager
from datetime import datetime
import os
DATABASE_URL = os.getenv(
"DATABASE_URL",
"postgresql+asyncpg://user:pass@localhost/db"
)
engine = create_async_engine(
DATABASE_URL,
echo=False,
pool_size=20,
max_overflow=10,
pool_pre_ping=True,
pool_recycle=3600,
)
AsyncSessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
autoflush=False,
)
Base = declarative_base()
class Developer(Base):
__tablename__ = "developers"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, nullable=False)
email = Column(String, unique=True, nullable=False, index=True)
specialty = Column(String, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
@asynccontextmanager
async def get_db_session():
async with AsyncSessionLocal() as session:
try:
async with session.begin():
yield session
await session.commit()
except Exception:
await session.rollback()
raise
async def get_developers_by_specialty(specialty: str) -> list[Developer]:
async with get_db_session() as session:
result = await session.execute(
select(Developer)
.where(Developer.specialty == specialty)
.order_by(Developer.name)
.limit(100)
)
return list(result.scalars().all())
async def bulk_create_developers(developers_data: list[dict]) -> int:
async with get_db_session() as session:
developers = [Developer(**data) for data in developers_data]
session.add_all(developers)
await session.flush()
count = len(developers)
return count
async def update_developer_email(developer_id: int, new_email: str) -> bool:
async with get_db_session() as session:
result = await session.execute(
select(Developer).where(Developer.id == developer_id)
)
developer = result.scalar_one_or_none()
if not developer:
return False
developer.email = new_email
developer.created_at = datetime.utcnow()
return TrueMore FastAPI Rules
FastAPI Dependency Injection Pattern
Use FastAPI's dependency injection system for authentication, database sessions, and shared business logic.
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session
from typing im...FastAPI WebSocket Real-time Patterns
Build real-time features with FastAPI WebSockets and connection manager patterns.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from typing import Dict,...FastAPI Background Tasks
Handle long-running operations with BackgroundTasks and Celery for distributed task queues.
from fastapi import FastAPI, BackgroundTasks, Depends, HTTPException
from pydantic import BaseModel, EmailStr
from sqlalchemy.orm import Session
from ...