FastAPI Async SQLAlchemy Fix
In this tutorial, you'll learn about FastAPI Async SQLAlchemy Fix. We cover key concepts, practical examples, and best practices.
The Problem
Default SQLAlchemy sessions are synchronous and block the async event loop. In FastAPI, blocking the event loop reduces throughput for all concurrent requests.
Quick Fix
Wrong — sync SQLAlchemy in async route
from sqlalchemy.orm import Session
@app.get("/items")
async def get_items(db: Session = Depends(get_db)):
# Blocks the event loop!
items = db.query(Item).all()
return items
Output: The database query blocks the event loop. Other requests queue up behind it.
Correct — async SQLAlchemy
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import select
# Use async driver (aiosqlite, asyncpg, aiomysql)
async_engine = create_async_engine(
"sqlite+aiosqlite:///./test.db",
echo=True,
)
AsyncSessionLocal = sessionmaker(
async_engine,
class_=AsyncSession,
expire_on_commit=False,
)
async def get_async_db():
async with AsyncSessionLocal() as session:
yield session
@app.get("/items")
async def get_items(db: AsyncSession = Depends(get_async_db)):
result = await db.execute(select(Item))
items = result.scalars().all()
return items
Output: Database queries are async. The event loop handles other requests during I/O wait.
Async CRUD operations
from sqlalchemy import select, update, delete
@app.post("/items")
async def create_item(name: str, db: AsyncSession = Depends(get_async_db)):
item = Item(name=name)
db.add(item)
await db.commit()
await db.refresh(item)
return item
@app.put("/items/{item_id}")
async def update_item(item_id: int, name: str, db: AsyncSession = Depends(get_async_db)):
await db.execute(
update(Item).where(Item.id == item_id).values(name=name)
)
await db.commit()
return {"status": "updated"}
@app.delete("/items/{item_id}")
async def delete_item(item_id: int, db: AsyncSession = Depends(get_async_db)):
await db.execute(delete(Item).where(Item.id == item_id))
await db.commit()
return {"status": "deleted"}
Async ORM relationships
@app.get("/categories/{cat_id}")
async def get_category(cat_id: int, db: AsyncSession = Depends(get_async_db)):
result = await db.execute(
select(Category).where(Category.id == cat_id)
)
category = result.scalar_one_or_none()
if category:
# Async load relationship
await db.refresh(category, ['items'])
return category
Run sync operations in thread pool
from asyncio import to_thread
@app.get("/legacy")
async def legacy_report():
# Run sync DB call in thread pool
result = await to_thread(run_sync_report)
return result
Prevention
- Use
AsyncSessionwith async database drivers (asyncpg, aiosqlite, aiomysql). - Use
await db.execute(select(...))instead ofdb.query(...). - Never use sync ORM session in async endpoints.
Common Mistakes with async sqlalchemy
- Using
returnto exit a function early instead of wrapping a pure value in the monad - Mixing let bindings with <- bindings in do notation, producing type errors
- Overlapping type class instances that cause GHC to reject the program with ambiguous dispatch errors
These mistakes appear frequently in real-world FASTAPI code. DodaTech's contributors have identified these patterns through analysis of open-source projects and production systems.
Practice Exercise
Write a pure function that safely divides two integers using Maybe, then test it with edge cases like division by zero and negative numbers.
This exercise reinforces the concepts covered in this guide. Try implementing it before checking online solutions.
FAQ
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro