How to Use SQLAlchemy with FastAPI: Simple Guide

To use SQLAlchemy with FastAPI, define your database models with SQLAlchemy ORM, create an async session with AsyncSession, and use dependency injection to access the session in your routes. This setup allows you to perform database operations asynchronously and cleanly within FastAPI endpoints.

Syntax

This is the basic pattern to use SQLAlchemy with FastAPI:

• Define your database URL and create an async engine.
• Set up a session maker with AsyncSession.
• Create your ORM models by inheriting from declarative_base().
• Use FastAPI dependencies to provide a session to your path operations.

python

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

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

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

Base = declarative_base()

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

app = FastAPI()

Example

This example shows a complete FastAPI app using SQLAlchemy async ORM to create and read users from a SQLite database.

python

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
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)
AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
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 AsyncSessionLocal() 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/{user_id}")
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)):
    result = await session.execute(select(User).where(User.id == user_id))
    user = result.scalars().first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "name": user.name}

Output

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

Common Pitfalls

Common mistakes when using SQLAlchemy with FastAPI include:

• Using synchronous SQLAlchemy sessions instead of async sessions, which blocks the event loop.
• Not using dependency injection for sessions, causing session management issues.
• Forgetting to commit transactions after adding or modifying data.
• Not creating database tables before running queries.

python

## Wrong: Using synchronous session in async route (blocks event loop)
# from sqlalchemy.orm import Session
# def get_session():
#     db = Session(bind=engine)
#     try:
#         yield db
#     finally:
#         db.close()

## Right: Use AsyncSession with async context
from sqlalchemy.ext.asyncio import AsyncSession
async def get_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

Quick Reference

Tips for using SQLAlchemy with FastAPI:

• Use create_async_engine and AsyncSession for async DB operations.
• Define models with declarative_base().
• Use FastAPI Depends to inject sessions into routes.
• Always commit and refresh after adding new objects.
• Run Base.metadata.create_all() on startup to create tables.

Key Takeaways

Use SQLAlchemy's async engine and AsyncSession for non-blocking database access in FastAPI.

Inject database sessions into routes using FastAPI's dependency system for clean session management.

Always commit transactions and refresh objects after adding or updating data.

Create database tables at app startup to avoid runtime errors.

Avoid mixing synchronous SQLAlchemy sessions in async FastAPI routes to prevent blocking.