SQLAlchemy

SQLAlchemy is the most widely used Python library for working with relational databases. It has two main parts:
1. SQLAlchemy Core — a SQL expression language for writing database queries in Python.
2. SQLAlchemy ORM — maps Python classes to database tables so you can read and write rows as Python objects instead of raw SQL.

Unlike Django, which ships with its own ORM, FastAPI has no built-in database layer. SQLAlchemy is the common choice for FastAPI projects.
Supported databases: SQLite, PostgreSQL, MySQL, MariaDB, Oracle, and others.

Install

$ pip install sqlalchemy

# For PostgreSQL
$ pip install psycopg2-binary

# For MySQL
$ pip install pymysql
      

Terms

Engine

The Engine is the starting point — it manages the connection to the database. You create it once with a database URL and reuse it across the app.

from sqlalchemy import create_engine

# SQLite (file-based, good for development)
engine = create_engine("sqlite:///./app.db", echo=True)

# PostgreSQL
# engine = create_engine("postgresql://user:password@localhost/mydb")
      

Model

A Model is a Python class that represents a database table. Each class attribute maps to a column. SQLAlchemy ORM uses a declarative base class to register models.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    email: Mapped[str] = mapped_column(unique=True)
      

Session

A Session is a workspace for database operations. You add, query, update, and delete objects through a session, then call commit() to save changes or rollback() to undo them.

from sqlalchemy.orm import Session

with Session(engine) as session:
    user = User(name="Amit", email="amit@example.com")
    session.add(user)
    session.commit()
      

How to Use (ORM Model)

Create Table

After defining models, create the tables in the database with Base.metadata.create_all().

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    email: Mapped[str] = mapped_column(unique=True)

engine = create_engine("sqlite:///./app.db")
Base.metadata.create_all(engine)   # Creates the users table
      

CRUD

Create, Read, Update, Delete — the four basic database operations:

from sqlalchemy.orm import Session
from sqlalchemy import select

with Session(engine) as session:
    # Create
    user = User(name="Amit", email="amit@example.com")
    session.add(user)
    session.commit()

    # Read
    users = session.scalars(select(User)).all()
    one_user = session.get(User, 1)          # by primary key

    # Update
    user = session.get(User, 1)
    user.name = "Amit Kumar"
    session.commit()

    # Delete
    session.delete(user)
    session.commit()
      

With FastAPI

In a FastAPI app, create the engine once and provide a database session per request using dependency injection:

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

def get_db():
    db = Session(engine)
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    return db.get(User, user_id)

@app.post("/users/")
def create_user(name: str, email: str, db: Session = Depends(get_db)):
    user = User(name=name, email=email)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user
      

Async SQLAlchemy

For async frameworks like FastAPI, SQLAlchemy provides async versions of the engine and session. Database I/O runs with await so other requests are not blocked while waiting for the database.
Install an async driver for your database:

# PostgreSQL
$ pip install asyncpg

# SQLite
$ pip install aiosqlite
      

create_async_engine()

create_async_engine() is the async equivalent of create_engine(). It creates an AsyncEngine that manages non-blocking connections to the database.
Use an async database URL (note the driver suffix +asyncpg or +aiosqlite):

from sqlalchemy.ext.asyncio import create_async_engine

# PostgreSQL (async)
engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/mydb",
    echo=True,
)

# SQLite (async)
# engine = create_async_engine("sqlite+aiosqlite:///./app.db")
      

async_sessionmaker()

async_sessionmaker() is the async equivalent of sessionmaker(). It is a factory that produces AsyncSession objects — use it with async with and await for ORM operations:

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:password@localhost/mydb")

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_users():
    async with AsyncSessionLocal() as session:
        result = await session.execute(select(User))
        return result.scalars().all()
      

Check Database Status

A common health-check pattern: open a connection, run a lightweight query (SELECT 1), and return True if the database responds or False if anything fails (network down, wrong credentials, database not running):

from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:password@localhost/mydb")

async def check_database() -> bool:
    try:
        async with engine.connect() as conn:
            await conn.execute(text("SELECT 1"))
        return True
    except Exception:
        return False

# Usage (e.g. FastAPI startup or /health endpoint)
# is_ok = await check_database()
      

How it works:
1. engine.connect() — opens an async connection from the pool.
2. text("SELECT 1") — runs a minimal query that every SQL database supports.
3. await conn.execute(...) — sends the query and waits for a response without blocking the event loop.
4. Returns False on any exception (connection refused, timeout, auth error, etc.).