How to Use Async SQLAlchemy with FastAPI: Simple Guide

To use async SQLAlchemy with FastAPI, set up an async engine with create_async_engine, use AsyncSession for database sessions, and run queries with await. Integrate these with FastAPI dependencies to handle async database calls cleanly.

Syntax

This is the basic syntax to set up async SQLAlchemy with FastAPI:

• create_async_engine: creates an async database engine.
• AsyncSession: manages async database sessions.
• async with: ensures proper session lifecycle.
• await session.execute(): runs async queries.
• Depends: injects session into FastAPI routes.

python

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

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

engine = create_async_engine(DATABASE_URL, echo=True)
async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)

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

app = FastAPI()

Example

This example shows a FastAPI app using async SQLAlchemy to add and read users from a SQLite database asynchronously.

python

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy import Column, Integer, String, select

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

engine = create_async_engine(DATABASE_URL, echo=False)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

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

app = FastAPI()

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

@app.post("/users/")
async def create_user(name: str, session: AsyncSession = Depends(get_session)):
    new_user = User(name=name)
    session.add(new_user)
    await session.commit()
    await session.refresh(new_user)
    return {"id": new_user.id, "name": new_user.name}

@app.get("/users/")
async def read_users(session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User))
    users = result.scalars().all()
    return users

Output

POST /users/ with JSON {"name": "Alice"} returns {"id": 1, "name": "Alice"}
GET /users/ returns [{"id": 1, "name": "Alice"}]

Common Pitfalls

Common mistakes when using async SQLAlchemy with FastAPI include:

• Using synchronous SQLAlchemy sessions instead of AsyncSession.
• Not using await with async database calls, causing errors or blocking.
• Not properly managing session lifecycle with async with or dependency injection.
• Mixing sync and async code in the same route without proper handling.

Always use AsyncSession and await for async queries, and manage sessions with FastAPI dependencies.

python

## Wrong way (sync session in async route):
# session = Session()  # synchronous session
# result = session.execute(select(User))  # blocking call

## Right way:
# async with async_session() as session:
#     result = await session.execute(select(User))

Quick Reference

Remember these key points when using async SQLAlchemy with FastAPI:

• Use create_async_engine to create the async engine.
• Use AsyncSession from sessionmaker with class_=AsyncSession.
• Use async with or FastAPI Depends to manage sessions.
• Always await async database calls.
• Run Base.metadata.create_all inside an async context on startup.

Key Takeaways

Use create_async_engine and AsyncSession for async database access in FastAPI.

Manage sessions with async context managers or FastAPI dependencies.

Always await async database calls to avoid blocking.

Initialize database tables asynchronously on app startup.

Avoid mixing sync SQLAlchemy sessions in async FastAPI routes.