SQLAlchemy 2.0: Complete Python ORM Guide

What is SQLAlchemy?

SQLAlchemy is Python’s most popular ORM, providing both high-level ORM patterns and low-level SQL expression language. Version 2.0 brings modern async support and improved type hints.

SQLAlchemy Features

Feature	Description
ORM	Object-relational mapping
Core	SQL expression language
Async	Full async/await support
Migrations	Alembic integration
Dialects	PostgreSQL, MySQL, SQLite

Getting Started

Installation

pip install sqlalchemy[asyncio]
pip install asyncpg  # PostgreSQL async
pip install psycopg2-binary  # PostgreSQL sync
pip install alembic  # Migrations

Basic Setup

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase

# Sync engine
engine = create_engine(
    "postgresql://user:password@localhost/dbname",
    echo=True  # Log SQL
)

# Session factory
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base class for models
class Base(DeclarativeBase):
    pass

Async Setup

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

# Async engine
async_engine = create_async_engine(
    "postgresql+asyncpg://user:password@localhost/dbname",
    echo=True
)

# Async session factory
AsyncSessionLocal = sessionmaker(
    async_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

Model Definition

Basic Models

from datetime import datetime
from typing import Optional, List
from sqlalchemy import String, Integer, Text, Boolean, DateTime, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(50))
    password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(default=True)
    created_at: Mapped[datetime] = mapped_column(default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        default=func.now(),
        onupdate=func.now()
    )

    # Relationships
    posts: Mapped[List["Post"]] = relationship(back_populates="author")
    comments: Mapped[List["Comment"]] = relationship(back_populates="user")

    def __repr__(self) -> str:
        return f"User(id={self.id}, email={self.email})"


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(255))
    slug: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    content: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    published: Mapped[bool] = mapped_column(default=False)
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(default=func.now())

    # Relationships
    author: Mapped["User"] = relationship(back_populates="posts")
    comments: Mapped[List["Comment"]] = relationship(
        back_populates="post",
        cascade="all, delete-orphan"
    )
    tags: Mapped[List["Tag"]] = relationship(
        secondary="post_tags",
        back_populates="posts"
    )


class Comment(Base):
    __tablename__ = "comments"

    id: Mapped[int] = mapped_column(primary_key=True)
    content: Mapped[str] = mapped_column(Text)
    post_id: Mapped[int] = mapped_column(ForeignKey("posts.id", ondelete="CASCADE"))
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(default=func.now())

    # Relationships
    post: Mapped["Post"] = relationship(back_populates="comments")
    user: Mapped["User"] = relationship(back_populates="comments")


class Tag(Base):
    __tablename__ = "tags"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50), unique=True)

    posts: Mapped[List["Post"]] = relationship(
        secondary="post_tags",
        back_populates="tags"
    )


# Many-to-many association table
from sqlalchemy import Table, Column

post_tags = Table(
    "post_tags",
    Base.metadata,
    Column("post_id", ForeignKey("posts.id"), primary_key=True),
    Column("tag_id", ForeignKey("tags.id"), primary_key=True)
)

Create Tables

# Sync
Base.metadata.create_all(engine)

# Async
async def create_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

CRUD Operations

Sync Operations

from sqlalchemy.orm import Session

def create_user(db: Session, email: str, username: str, password: str) -> User:
    user = User(email=email, username=username, password=password)
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

def get_user(db: Session, user_id: int) -> Optional[User]:
    return db.get(User, user_id)

def get_user_by_email(db: Session, email: str) -> Optional[User]:
    return db.query(User).filter(User.email == email).first()

def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
    return db.query(User).offset(skip).limit(limit).all()

def update_user(db: Session, user_id: int, **kwargs) -> Optional[User]:
    user = db.get(User, user_id)
    if user:
        for key, value in kwargs.items():
            setattr(user, key, value)
        db.commit()
        db.refresh(user)
    return user

def delete_user(db: Session, user_id: int) -> bool:
    user = db.get(User, user_id)
    if user:
        db.delete(user)
        db.commit()
        return True
    return False

# Usage
with SessionLocal() as db:
    user = create_user(db, "john@example.com", "johndoe", "hashedpass")
    print(user)

Async Operations

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

async def create_user(db: AsyncSession, email: str, username: str, password: str) -> User:
    user = User(email=email, username=username, password=password)
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def get_user(db: AsyncSession, user_id: int) -> Optional[User]:
    return await db.get(User, user_id)

async def get_user_by_email(db: AsyncSession, email: str) -> Optional[User]:
    result = await db.execute(
        select(User).where(User.email == email)
    )
    return result.scalar_one_or_none()

async def get_users(db: AsyncSession, skip: int = 0, limit: int = 100) -> List[User]:
    result = await db.execute(
        select(User).offset(skip).limit(limit)
    )
    return result.scalars().all()

async def update_user(db: AsyncSession, user_id: int, **kwargs) -> Optional[User]:
    user = await db.get(User, user_id)
    if user:
        for key, value in kwargs.items():
            setattr(user, key, value)
        await db.commit()
        await db.refresh(user)
    return user

async def delete_user(db: AsyncSession, user_id: int) -> bool:
    user = await db.get(User, user_id)
    if user:
        await db.delete(user)
        await db.commit()
        return True
    return False

# Usage
async with AsyncSessionLocal() as db:
    user = await create_user(db, "john@example.com", "johndoe", "hashedpass")

Advanced Queries

Select Statements

from sqlalchemy import select, and_, or_, not_, func, desc, asc

# Basic select
stmt = select(User)
result = await db.execute(stmt)
users = result.scalars().all()

# Select specific columns
stmt = select(User.id, User.email)
result = await db.execute(stmt)
rows = result.all()

# Filtering
stmt = select(User).where(User.is_active == True)

# Multiple conditions
stmt = select(Post).where(
    and_(
        Post.published == True,
        Post.created_at > datetime(2024, 1, 1),
        or_(
            Post.title.contains("Python"),
            Post.title.contains("SQLAlchemy")
        )
    )
)

# Like / ilike
stmt = select(Post).where(Post.title.ilike("%tutorial%"))

# In clause
stmt = select(User).where(User.id.in_([1, 2, 3, 4, 5]))

# Is null
stmt = select(Post).where(Post.content.is_(None))

# Order by
stmt = select(Post).order_by(desc(Post.created_at), asc(Post.title))

# Limit and offset
stmt = select(Post).limit(10).offset(20)

# Distinct
stmt = select(Post.author_id).distinct()

Joins

from sqlalchemy.orm import joinedload, selectinload

# Inner join
stmt = select(Post, User).join(User, Post.author_id == User.id)
result = await db.execute(stmt)

# Left join
stmt = select(Post, User).join(User, Post.author_id == User.id, isouter=True)

# Eager loading - joined load
stmt = select(Post).options(joinedload(Post.author)).where(Post.published == True)
result = await db.execute(stmt)
posts = result.scalars().unique().all()

# Eager loading - selectin load (better for collections)
stmt = select(User).options(selectinload(User.posts))
result = await db.execute(stmt)
users = result.scalars().all()

# Multiple eager loads
stmt = select(Post).options(
    joinedload(Post.author),
    selectinload(Post.comments).joinedload(Comment.user),
    selectinload(Post.tags)
)

Aggregations

from sqlalchemy import func

# Count
stmt = select(func.count()).select_from(User)
result = await db.execute(stmt)
count = result.scalar()

# Count with filter
stmt = select(func.count()).select_from(User).where(User.is_active == True)

# Group by
stmt = select(
    Post.author_id,
    func.count(Post.id).label("post_count")
).group_by(Post.author_id)

# Having
stmt = select(
    Post.author_id,
    func.count(Post.id).label("post_count")
).group_by(Post.author_id).having(func.count(Post.id) > 5)

# Aggregate functions
stmt = select(
    func.min(Post.created_at),
    func.max(Post.created_at),
    func.avg(func.length(Post.content))
)

Subqueries

# Subquery
subq = select(func.count(Post.id)).where(Post.author_id == User.id).scalar_subquery()
stmt = select(User, subq.label("post_count"))

# Exists
from sqlalchemy import exists

stmt = select(User).where(
    exists(select(Post).where(Post.author_id == User.id))
)

Transactions

# Sync transaction
with SessionLocal() as db:
    try:
        user = User(email="test@example.com", username="test", password="pass")
        db.add(user)

        post = Post(title="Test", slug="test", author=user)
        db.add(post)

        db.commit()
    except Exception:
        db.rollback()
        raise

# Async transaction
async with AsyncSessionLocal() as db:
    async with db.begin():
        user = User(email="test@example.com", username="test", password="pass")
        db.add(user)

        post = Post(title="Test", slug="test", author=user)
        db.add(post)
        # Auto-commits on exit, auto-rollback on exception

# Nested transactions (savepoints)
async with AsyncSessionLocal() as db:
    async with db.begin():
        user = User(email="test@example.com", username="test", password="pass")
        db.add(user)

        async with db.begin_nested():
            # This can be rolled back independently
            post = Post(title="Test", slug="test", author=user)
            db.add(post)

FastAPI Integration

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from contextlib import asynccontextmanager

app = FastAPI()

# Dependency
async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.post("/users/", response_model=UserSchema)
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db)
):
    # Check if user exists
    existing = await get_user_by_email(db, user_data.email)
    if existing:
        raise HTTPException(400, "Email already registered")

    return await create_user(db, **user_data.dict())

@app.get("/users/{user_id}", response_model=UserSchema)
async def read_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await get_user(db, user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

@app.get("/posts/", response_model=List[PostSchema])
async def read_posts(
    skip: int = 0,
    limit: int = 10,
    db: AsyncSession = Depends(get_db)
):
    stmt = select(Post).options(
        joinedload(Post.author)
    ).where(Post.published == True).offset(skip).limit(limit)

    result = await db.execute(stmt)
    return result.scalars().unique().all()

Migrations with Alembic

# Initialize Alembic
alembic init alembic

# Generate migration
alembic revision --autogenerate -m "create users table"

# Apply migrations
alembic upgrade head

# Rollback
alembic downgrade -1

alembic.ini

[alembic]
script_location = alembic
sqlalchemy.url = postgresql://user:password@localhost/dbname

env.py

from alembic import context
from sqlalchemy import engine_from_config, pool
from models import Base

target_metadata = Base.metadata

def run_migrations_online():
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()

Summary

Operation	SQLAlchemy 2.0 Syntax
Select	select(Model).where()
Insert	db.add(instance)
Update	setattr() + commit()
Delete	db.delete(instance)
Join	select().join()
Eager Load	.options(joinedload())
Transaction	async with db.begin()

SQLAlchemy 2.0 provides powerful, type-safe database operations with excellent async support for modern Python applications.