Tutorial: Learn Python ORM with SQLAlchemy from Scratch (2026)
I learned SQLAlchemy the hard way — by debugging N+1 queries in production at 2 AM. SQLAlchemy is Python's most powerful ORM (Object-Relational Mapper). It lets you work with databases using Python objects instead of raw SQL. The 2.0 version (current in 2026) introduced a streamlined API with better type hints, clearer query patterns, and native async support. Unlike Django's ORM, SQLAlchemy is framework-agnostic and works with Flask, FastAPI, or standalone scripts.
This tutorial covers SQLAlchemy 2.0 from models to migrations: defining tables as Python classes, querying with the modern select() API, filtering and joining, relationships and eager loading, and managing schema changes with Alembic. We'll model a blog with users, posts, and comments — the standard 'hello world' of ORMs.
Models and Engine Configuration
SQLAlchemy models inherit from DeclarativeBase (2.0 style) or use the older declarative_base(). Each model attribute maps to a database column with a type. The Engine manages the database connection pool and speaks SQL dialect. Create the engine with create_engine() for sync or create_async_engine() for async. Metadata from models is used to create tables with Base.metadata.create_all(engine).
from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.orm import DeclarativeBase
from datetime import datetime
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True, nullable=False)
email = Column(String(120), unique=True, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
engine = create_engine('sqlite:///blog.db', echo=True)
Base.metadata.create_all(engine)Sessions and CRUD Operations
The Session is the workspace for database operations. Create it with sessionmaker bound to the engine. Sessions manage transactions — use session.add() to insert, session.commit() to persist, and session.rollback() to revert. Each session tracks object changes and auto-flushes before queries. The context manager pattern (with Session() as session) ensures proper cleanup.
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
# Create
with Session() as session:
user = User(username='alice', email='alice@example.com')
session.add(user)
session.commit()
print(f"Created user {user.id}")
# Read
with Session() as session:
user = session.get(User, 1)
print(user.username, user.email)
# Update
with Session() as session:
user = session.get(User, 1)
user.email = 'alice@newdomain.com'
session.commit()
# Delete
with Session() as session:
user = session.get(User, 1)
session.delete(user)
session.commit()Querying with the 2.0 Select API
SQLAlchemy 2.0 introduced the select() function as the primary querying interface. Use select(Model).where(conditions) to build queries. The session executes with session.execute(statement).scalars() returns model instances (not tuples). Chaining methods like .where(), .order_by(), .limit(), and .offset() builds complex queries. The .where() accepts multiple conditions joined by AND.
from sqlalchemy import select
with Session() as session:
# Get all users
stmt = select(User)
users = session.execute(stmt).scalars().all()
# Filter with conditions
stmt = select(User).where(User.username == 'alice')
user = session.execute(stmt).scalars().one()
# Multiple conditions and ordering
stmt = select(User).where(
User.email.like('%@example.com'),
User.created_at > datetime(2025, 1, 1)
).order_by(User.created_at.desc()).limit(10)
results = session.execute(stmt).scalars().all()
# Count
stmt = select(User).where(User.username.like('a%'))
count = session.execute(stmt).scalars().count()
print(f"Users starting with 'a': {count}")Relationships: One-to-Many and Many-to-Many
Relationships define how models connect. A ForeignKey column in the child table references the parent's primary key. The relationship() directive creates Python-side navigation (user.posts, post.author). Lazy loading (default) fetches related objects on first access. Eager loading with .options(selectinload(Relationship)) avoids N+1 queries by fetching related data in a single query.
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship, selectinload
class Post(Base):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True)
title = Column(String(200), nullable=False)
body = Column(Text)
user_id = Column(Integer, ForeignKey('users.id'))
created_at = Column(DateTime, default=datetime.utcnow)
author = relationship('User', back_populates='posts')
User.posts = relationship('Post', back_populates='author', cascade='all, delete-orphan')
# Eager loading to avoid N+1
with Session() as session:
stmt = select(User).options(selectinload(User.posts))
users = session.execute(stmt).scalars().all()
for user in users:
print(f"{user.username}: {len(user.posts)} posts")
for post in user.posts:
print(f" - {post.title}")Migrations with Alembic
Alembic manages database schema migrations — version-controlled changes to tables, columns, and indexes. Initialize with alembic init alembic, configure the database URL in alembic.ini, and run alembic revision --autogenerate -m 'message' to detect model changes. Alembic generates migration scripts showing the upgrade and downgrade paths. Apply with alembic upgrade head.
# Terminal commands
# alembic init alembic
# Edit alembic.ini: sqlalchemy.url = sqlite:///blog.db
# In alembic/env.py: from models import Base; target_metadata = Base.metadata
# Generate a migration
alembic revision --autogenerate -m "add comment model"
# Review the generated script (in alembic/versions/)
# Apply migration
alembic upgrade head
# Rollback one step
alembic downgrade -1
# Generated migration example:
"""
def upgrade():
op.create_table('comments',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('body', sa.Text(), nullable=True),
sa.Column('post_id', sa.Integer(), nullable=True),
sa.Column('user_id', sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(['post_id'], ['posts.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
"""Async SQLAlchemy with Asyncio
SQLAlchemy 2.0 supports async drivers (asyncpg for PostgreSQL, aiosqlite for SQLite). Use create_async_engine and AsyncSession from sqlalchemy.ext.asyncio. The query API is identical to sync, but you await execute(), commit(), and close(). Async sessions must be used with async with or explicit await management. This pairs perfectly with FastAPI and other async frameworks.
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
import asyncio
DATABASE_URL = "sqlite+aiosqlite:///./blog.db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession)
async def create_user():
async with AsyncSessionLocal() as session:
user = User(username='bob', email='bob@example.com')
session.add(user)
await session.commit()
print(f"Created user {user.id}")
async def get_users():
async with AsyncSessionLocal() as session:
stmt = select(User)
result = await session.execute(stmt)
users = result.scalars().all()
return users
asyncio.run(create_user())
users = asyncio.run(get_users())
print(users)Frequently Asked Questions
Should I use SQLAlchemy or raw SQL?
SQLAlchemy adds abstraction that helps with maintainability, migration management, and database portability. Raw SQL gives full control and sometimes better performance. I use SQLAlchemy for most applications and raw SQL for complex reporting queries.
How do I avoid the N+1 query problem?
Use eager loading with selectinload() or joinedload() in your query options. Monitor emitted SQL with echo=True and review logs. Tools like sqlalchemy-toolbar or Flask-DebugToolbar highlight N+1 issues.
What's the difference between SQLAlchemy 1.x and 2.0?
2.0 is a major API cleanup: unified query pattern with select(), native async support, better type annotations, and removal of deprecated patterns (Session.query()). All new projects should use 2.0 style.
How do I handle database migrations in production?
Use Alembic with version-controlled migration scripts. Run alembic upgrade head as part of your deployment process. Always test migrations on a staging database first. Backup before running destructive migrations.
Originally published on Ayodhyyya. Last updated June 1, 2026.