How To Use SQLAlchemy 2.0 ORM with Python

Intermediate

SQLAlchemy is the gold standard for Object-Relational Mapping in Python. Version 2.0 represents a major evolution, introducing a more intuitive API that emphasizes explicit, modern patterns while maintaining backward compatibility. Whether you’re building a small Flask application or a complex data management system, SQLAlchemy 2.0 provides the tools to interact with databases using Python objects instead of raw SQL strings.

The ORM (Object-Relational Mapping) layer in SQLAlchemy 2.0 allows you to define database tables as Python classes, called models. Once you define a model, you can perform all database operations–creating records, querying data, updating rows, and deleting entries–using Pythonic syntax. The new select() construct and DeclarativeBase provide clearer, more expressive patterns than earlier versions.

In this tutorial, we’ll explore the key features of SQLAlchemy 2.0 ORM: how to define models, manage database sessions, perform CRUD operations, query data with the new select() API, establish relationships between tables, handle transactions, and build a real-world example. By the end, you’ll understand how to leverage SQLAlchemy 2.0 to create robust, maintainable database-driven applications.

Quick Example: 20 Lines of SQLAlchemy 2.0

Let’s start with a complete, working example to see SQLAlchemy 2.0 in action:

# quick_example.py
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Session, 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(String(50))

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(User(name='Alice'))
    session.add(User(name='Bob'))
    session.commit()

with Session(engine) as session:
    from sqlalchemy import select
    users = session.scalars(select(User)).all()
    for user in users:
        print(f'{user.id}: {user.name}')

Output:

1: Alice
2: Bob

This example demonstrates the core workflow: define a model inheriting from DeclarativeBase, create an engine, manage a session, insert records, and query them using select(). Notice the type hints (Mapped[str]) and the modern syntax–this is SQLAlchemy 2.0 style.

What Is SQLAlchemy ORM?

SQLAlchemy provides multiple ways to interact with databases. The ORM layer sits at the highest abstraction level, letting you work with Python objects. Here’s how it compares to alternatives:

Approach	How It Works	Pros	Cons
Raw SQL	Write SQL strings directly in Python	Maximum control, direct database access	Error-prone, requires manual parameter binding, not Pythonic
SQLAlchemy Core	Use SQL expression language to build queries programmatically	Type-safe, database-agnostic, composable	Still working with table/column constructs, not Python objects
SQLAlchemy ORM	Map database tables to Python classes, query objects directly	Pythonic, intuitive, supports relationships and complex queries, automatic change tracking	Slightly more overhead, must understand session lifecycle

SQLAlchemy 2.0’s ORM is the most productive choice for most applications because it combines clarity with power. You define your data model once, and the ORM handles the translation to SQL behind the scenes.

Installing SQLAlchemy and Setting Up

Install SQLAlchemy using pip:

# shell
pip install sqlalchemy

Output:

Successfully installed sqlalchemy-2.0.x

Verify the installation:

# check_version.py
import sqlalchemy
print(f'SQLAlchemy version: {sqlalchemy.__version__}')

Output:

SQLAlchemy version: 2.0.x

For this tutorial, we’ll use SQLite in-memory databases (specified as sqlite:///:memory:), which requires no external setup. For production use with PostgreSQL, MySQL, or other databases, install the appropriate driver (e.g., pip install psycopg2-binary for PostgreSQL).

Defining Models with DeclarativeBase

In SQLAlchemy 2.0, you define models by creating a class that inherits from DeclarativeBase. This base class automatically handles the mapping between your Python class and the database table.

Creating the DeclarativeBase

# models_setup.py
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

Output:

(No output - this defines the base class)

The Base class is the foundation for all your models. It tracks metadata (table definitions) and provides utilities for creating tables.

Defining a Model with Columns

Use Mapped and mapped_column() to define model attributes in SQLAlchemy 2.0:

# product_model.py
from sqlalchemy import String, Float, Integer
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Product(Base):
    __tablename__ = 'products'

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    price: Mapped[float] = mapped_column(Float, default=0.0)
    stock: Mapped[int] = mapped_column(Integer, default=0)

Output:

(No output - this defines the model structure)

Key points about this model:

• __tablename__ specifies the database table name
• Mapped[type] is a type hint that declares the Python type of the column
• mapped_column() specifies database-level constraints (primary key, nullability, defaults)
• primary_key=True makes id the primary key with auto-increment behavior
• nullable=False ensures the name field cannot be NULL
• default=0.0 provides a default value for new records

Common Column Types

# column_types_example.py
from sqlalchemy import String, Integer, Float, Boolean, DateTime, Text, Date
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime, date

class Base(DeclarativeBase):
    pass

class Article(Base):
    __tablename__ = 'articles'

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(255))
    content: Mapped[str] = mapped_column(Text)
    is_published: Mapped[bool] = mapped_column(Boolean, default=False)
    rating: Mapped[float] = mapped_column(Float)
    views: Mapped[int] = mapped_column(Integer, default=0)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
    published_date: Mapped[date] = mapped_column(Date, nullable=True)

Output:

(No output - demonstrates various column types)

Creating Tables and the Engine

The SQLAlchemy engine is your gateway to the database. It manages connections and executes SQL. To create tables, you call metadata.create_all():

# create_tables.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)

# Create an in-memory SQLite database
engine = create_engine('sqlite:///:memory:', echo=False)

# Create all tables defined in metadata
Base.metadata.create_all(engine)

print('Tables created successfully!')

Output:

Tables created successfully!

The connection string format is dialect+driver://user:password@host:port/database. Examples:

• sqlite:///:memory: – In-memory SQLite (perfect for testing)
• sqlite:///app.db – File-based SQLite
• postgresql://user:pass@localhost/dbname – PostgreSQL
• mysql+pymysql://user:pass@localhost/dbname – MySQL

Sessions and Basic CRUD Operations

A Session is a context manager that tracks changes to your objects and coordinates with the database. CRUD stands for Create, Read, Update, Delete–the fundamental database operations.

Create (Insert) Records

# create_records.py
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Book(Base):
    __tablename__ = 'books'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    author: Mapped[str] = mapped_column(String(100))

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# Create and insert records
with Session(engine) as session:
    book1 = Book(title='Python Basics', author='Alice Johnson')
    book2 = Book(title='Web Dev with Django', author='Bob Smith')

    session.add(book1)
    session.add(book2)
    session.commit()

    print(f'Created book with ID: {book1.id}')
    print(f'Created book with ID: {book2.id}')

Output:

Created book with ID: 1
Created book with ID: 2

When you commit, SQLAlchemy assigns primary keys (IDs) to new objects. The session tracks the objects and only issues SQL when you call commit().

Read (Query) Records

# read_records.py
from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Book(Base):
    __tablename__ = 'books'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    author: Mapped[str] = mapped_column(String(100))

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(Book(title='Python Basics', author='Alice Johnson'))
    session.add(Book(title='Web Dev with Django', author='Bob Smith'))
    session.commit()

# Read records
with Session(engine) as session:
    stmt = select(Book)
    books = session.scalars(stmt).all()

    for book in books:
        print(f'{book.id}: {book.title} by {book.author}')

Output:

1: Python Basics by Alice Johnson
2: Web Dev with Django by Bob Smith

Update Records

# update_records.py
from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Book(Base):
    __tablename__ = 'books'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    author: Mapped[str] = mapped_column(String(100))

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(Book(title='Python Basics', author='Alice Johnson'))
    session.commit()

# Update a record
with Session(engine) as session:
    stmt = select(Book).where(Book.title == 'Python Basics')
    book = session.scalars(stmt).first()

    if book:
        book.author = 'Alice J. Johnson'
        session.commit()
        print(f'Updated: {book.title} by {book.author}')

Output:

Updated: Python Basics by Alice J. Johnson

Delete Records

# delete_records.py
from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Book(Base):
    __tablename__ = 'books'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(Book(title='Old Book'))
    session.commit()

# Delete a record
with Session(engine) as session:
    stmt = select(Book).where(Book.title == 'Old Book')
    book = session.scalars(stmt).first()

    if book:
        session.delete(book)
        session.commit()
        print('Book deleted successfully')

Output:

Book deleted successfully

Querying with select()

SQLAlchemy 2.0’s select() construct is the modern way to build queries. It’s more expressive than the legacy query() method and provides better IDE support through type hints.

Basic Selects

# basic_select.py
from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Student(Base):
    __tablename__ = 'students'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    grade: Mapped[int]

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(Student(name='Alice', grade=95))
    session.add(Student(name='Bob', grade=87))
    session.add(Student(name='Charlie', grade=92))
    session.commit()

# Select all
with Session(engine) as session:
    stmt = select(Student)
    all_students = session.scalars(stmt).all()
    print(f'Total students: {len(all_students)}')

    # Select first
    first = session.scalars(select(Student)).first()
    print(f'First student: {first.name}')

Output:

Total students: 3
First student: Alice

Filtering Results

# filtering.py
from sqlalchemy import create_engine, String, select
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Student(Base):
    __tablename__ = 'students'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    grade: Mapped[int]

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add_all([
        Student(name='Alice', grade=95),
        Student(name='Bob', grade=87),
        Student(name='Charlie', grade=92),
        Student(name='Diana', grade=88)
    ])
    session.commit()

with Session(engine) as session:
    # Equal comparison
    stmt = select(Student).where(Student.name == 'Alice')
    result = session.scalars(stmt).first()
    print(f'Found: {result.name} (Grade: {result.grade})')

    # Greater than
    stmt = select(Student).where(Student.grade > 90)
    high_performers = session.scalars(stmt).all()
    print(f'High performers: {[s.name for s in high_performers]}')

    # Like pattern
    stmt = select(Student).where(Student.name.like('D%'))
    result = session.scalars(stmt).first()
    print(f'Names starting with D: {result.name}')

Output:

Found: Alice (Grade: 95)
High performers: ['Alice', 'Charlie']
Names starting with D: Diana

Ordering and Limiting

# ordering_limiting.py
from sqlalchemy import create_engine, String, select, desc
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Student(Base):
    __tablename__ = 'students'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    grade: Mapped[int]

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add_all([
        Student(name='Alice', grade=95),
        Student(name='Bob', grade=87),
        Student(name='Charlie', grade=92),
        Student(name='Diana', grade=88)
    ])
    session.commit()

with Session(engine) as session:
    # Order ascending
    stmt = select(Student).order_by(Student.grade)
    lowest = session.scalars(stmt).first()
    print(f'Lowest grade: {lowest.name} ({lowest.grade})')

    # Order descending
    stmt = select(Student).order_by(desc(Student.grade))
    highest = session.scalars(stmt).first()
    print(f'Highest grade: {highest.name} ({highest.grade})')

    # Limit
    stmt = select(Student).order_by(desc(Student.grade)).limit(2)
    top_two = session.scalars(stmt).all()
    print(f'Top 2 students: {[s.name for s in top_two]}')

Output:

Lowest grade: Bob (87)
Highest grade: Alice (95)
Top 2 students: ['Alice', 'Charlie']

Joins Between Tables

# joins.py
from sqlalchemy import create_engine, String, select, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Department(Base):
    __tablename__ = 'departments'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    employees: Mapped[list['Employee']] = relationship(back_populates='department')

class Employee(Base):
    __tablename__ = 'employees'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    department_id: Mapped[int] = mapped_column(ForeignKey('departments.id'))
    department: Mapped[Department] = relationship(back_populates='employees')

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    dept_eng = Department(name='Engineering')
    dept_hr = Department(name='HR')

    session.add_all([
        Employee(name='Alice', department=dept_eng),
        Employee(name='Bob', department=dept_eng),
        Employee(name='Charlie', department=dept_hr)
    ])
    session.commit()

with Session(engine) as session:
    # Join departments and employees
    stmt = select(Employee).join(Department).where(Department.name == 'Engineering')
    eng_employees = session.scalars(stmt).all()
    print(f'Engineering employees: {[e.name for e in eng_employees]}')

Output:

Engineering employees: ['Alice', 'Bob']

Relationships Between Models

Relationships let you traverse from one model to another. SQLAlchemy handles the foreign key constraints and makes it easy to load related objects.

One-to-Many Relationships

# one_to_many.py
from sqlalchemy import create_engine, String, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Author(Base):
    __tablename__ = 'authors'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    books: Mapped[list['Book']] = relationship(back_populates='author', cascade='all, delete-orphan')

class Book(Base):
    __tablename__ = 'books'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(100))
    author_id: Mapped[int] = mapped_column(ForeignKey('authors.id'))
    author: Mapped[Author] = relationship(back_populates='books')

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    author = Author(name='George Orwell')
    author.books = [
        Book(title='1984'),
        Book(title='Animal Farm')
    ]
    session.add(author)
    session.commit()

with Session(engine) as session:
    from sqlalchemy import select
    author = session.scalars(select(Author).where(Author.name == 'George Orwell')).first()
    print(f'Author: {author.name}')
    for book in author.books:
        print(f'  - {book.title}')

Output:

Author: George Orwell
  - 1984
  - Animal Farm

Many-to-Many Relationships

# many_to_many.py
from sqlalchemy import create_engine, String, ForeignKey, Table, Column
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

# Association table for many-to-many
student_course = Table(
    'student_course',
    Base.metadata,
    Column('student_id', ForeignKey('students.id'), primary_key=True),
    Column('course_id', ForeignKey('courses.id'), primary_key=True)
)

class Student(Base):
    __tablename__ = 'students'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    courses: Mapped[list['Course']] = relationship(secondary=student_course, back_populates='students')

class Course(Base):
    __tablename__ = 'courses'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    students: Mapped[list[Student]] = relationship(secondary=student_course, back_populates='courses')

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    python = Course(name='Python 101')
    math = Course(name='Calculus I')

    alice = Student(name='Alice', courses=[python, math])
    bob = Student(name='Bob', courses=[python])

    session.add_all([alice, bob])
    session.commit()

with Session(engine) as session:
    from sqlalchemy import select
    student = session.scalars(select(Student).where(Student.name == 'Alice')).first()
    print(f'Alice is taking: {[c.name for c in student.courses]}')

Output:

Alice is taking: ['Python 101', 'Calculus I']

Transactions and Error Handling

A transaction is a sequence of database operations that either all succeed or all fail. SQLAlchemy sessions handle transactions automatically, but you can control commit/rollback behavior explicitly.

Basic Commit and Rollback

# transactions.py
from sqlalchemy import create_engine, String
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class Account(Base):
    __tablename__ = 'accounts'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    balance: Mapped[float]

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# Create initial accounts
with Session(engine) as session:
    session.add_all([
        Account(name='Alice', balance=1000.0),
        Account(name='Bob', balance=500.0)
    ])
    session.commit()

# Simulate a transfer with error handling
with Session(engine) as session:
    try:
        alice = session.query(Account).filter_by(name='Alice').first()
        bob = session.query(Account).filter_by(name='Bob').first()

        # Transfer 200 from Alice to Bob
        alice.balance -= 200
        bob.balance += 200

        session.commit()
        print(f'Transfer successful: Alice={alice.balance}, Bob={bob.balance}')
    except Exception as e:
        session.rollback()
        print(f'Transfer failed: {e}')

Output:

Transfer successful: Alice=800.0, Bob=700.0

Error Handling with Try-Except

# error_handling.py
from sqlalchemy import create_engine, String, exc
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = 'users'
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(100), unique=True)

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add(User(email='alice@example.com'))
    session.commit()

# Try to add duplicate email
with Session(engine) as session:
    try:
        session.add(User(email='alice@example.com'))
        session.commit()
    except exc.IntegrityError as e:
        session.rollback()
        print('Error: Email already exists')
    except Exception as e:
        session.rollback()
        print(f'Unexpected error: {e}')

Output:

Error: Email already exists

Real-Life Example: Blog Database

Let’s build a complete blog system with Post, Author, and Tag models, demonstrating relationships, CRUD operations, and queries.

# blog_system.py
from sqlalchemy import create_engine, String, Text, ForeignKey, Table, Column, select, desc
from sqlalchemy.orm import DeclarativeBase, Session, Mapped, mapped_column, relationship
from datetime import datetime

class Base(DeclarativeBase):
    pass

# Association table for many-to-many relationship
post_tag = Table(
    'post_tag',
    Base.metadata,
    Column('post_id', ForeignKey('posts.id'), primary_key=True),
    Column('tag_id', ForeignKey('tags.id'), primary_key=True)
)

class Author(Base):
    __tablename__ = 'authors'
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)
    email: Mapped[str] = mapped_column(String(100), unique=True)
    posts: Mapped[list['Post']] = relationship(back_populates='author', cascade='all, delete-orphan')

class Post(Base):
    __tablename__ = 'posts'
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200), nullable=False)
    content: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)
    author_id: Mapped[int] = mapped_column(ForeignKey('authors.id'))
    author: Mapped[Author] = relationship(back_populates='posts')
    tags: Mapped[list['Tag']] = relationship(secondary=post_tag, back_populates='posts')

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_tag, back_populates='tags')

# Setup
engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

# Create blog data
with Session(engine) as session:
    author1 = Author(name='Alice', email='alice@blog.com')
    author2 = Author(name='Bob', email='bob@blog.com')

    python_tag = Tag(name='Python')
    web_tag = Tag(name='Web')

    post1 = Post(
        title='Getting Started with Python',
        content='Python is a great language...',
        author=author1,
        tags=[python_tag, web_tag]
    )
    post2 = Post(
        title='Advanced ORM Techniques',
        content='SQLAlchemy provides powerful ORM features...',
        author=author1,
        tags=[python_tag]
    )
    post3 = Post(
        title='Web Development Tips',
        content='Here are some web development best practices...',
        author=author2,
        tags=[web_tag]
    )

    session.add_all([author1, author2, python_tag, web_tag, post1, post2, post3])
    session.commit()

# Query examples
with Session(engine) as session:
    # Find all posts by an author
    stmt = select(Post).join(Author).where(Author.name == 'Alice').order_by(desc(Post.created_at))
    alice_posts = session.scalars(stmt).all()
    print(f'Posts by Alice: {len(alice_posts)}')
    for post in alice_posts:
        print(f'  - {post.title}')

    # Find all posts with a specific tag
    stmt = select(Post).join(Post.tags).where(Tag.name == 'Python')
    python_posts = session.scalars(stmt).all()
    print(f'\nPython posts: {len(python_posts)}')
    for post in python_posts:
        print(f'  - {post.title} by {post.author.name}')

    # Count total posts
    stmt = select(Post)
    total_posts = len(session.scalars(stmt).all())
    print(f'\nTotal blog posts: {total_posts}')

Output:

Posts by Alice: 2
  - Advanced ORM Techniques
  - Getting Started with Python

Python posts: 2
  - Getting Started with Python by Alice
  - Advanced ORM Techniques by Alice

Total blog posts: 3

This example showcases the full power of SQLAlchemy 2.0 ORM: defining multiple related models, using many-to-many relationships, performing complex queries with joins, and maintaining referential integrity through cascading deletes.

Frequently Asked Questions

What’s the difference between Mapped and traditional type hints?

Mapped[type] is SQLAlchemy 2.0’s way to combine Python type hints with ORM metadata. It tells SQLAlchemy about the column while also providing type information to your IDE and type checkers. The legacy approach used no type hints.

When should I use relationships versus manual joins?

Use relationships when you want to access related objects as Python attributes (e.g., author.posts). Use manual joins when you need more control over the query or want to fetch only specific columns. Relationships are more Pythonic and handle lazy loading by default.

What’s the difference between add() and add_all()?

session.add(obj) adds a single object. session.add_all([obj1, obj2]) adds multiple objects at once. Use add_all() for convenience when inserting several objects.

How do I handle database connection pooling?

SQLAlchemy’s engine manages connection pooling automatically. For production applications, configure pool settings when creating the engine: engine = create_engine('postgresql://...', pool_size=10, max_overflow=20).

Can I use SQLAlchemy ORM with async code?

Yes! SQLAlchemy 2.0 includes async support using AsyncSession and create_async_engine(). This is useful for high-concurrency web applications. However, the basic patterns remain the same.

What happens if I forget to commit()?

Changes are held in the session but not persisted to the database. When the session context exits (the with block ends), uncommitted changes are rolled back. Always call commit() to save changes.

How do I avoid the N+1 query problem?

The N+1 problem happens when loading a parent object triggers a separate query for each child. Use eager loading with selectinload() or joinedload() to fetch related objects in one query: select(Author).options(selectinload(Author.posts)).

Conclusion

SQLAlchemy 2.0 brings modern Python patterns to database programming. By using DeclarativeBase for model definition, select() for queries, and proper session management, you can build robust data-driven applications without writing a single SQL string. The ORM layer abstracts away database details while remaining transparent and powerful.

Key takeaways from this tutorial:

• Models inherit from DeclarativeBase and use Mapped type hints
• select() is the modern way to build type-safe queries
• Sessions manage transactions and object tracking
• Relationships make it natural to traverse related objects
• Always handle errors and rollback on failure
• Eager loading prevents common performance pitfalls

For next steps, explore SQLAlchemy’s advanced features like hybrid properties, custom types, and query optimizations. Consider integrating SQLAlchemy with frameworks like Flask or FastAPI for web development. As you grow more comfortable with the ORM, you’ll find that SQLAlchemy’s power and flexibility make it an excellent choice for any Python project requiring database interaction.

Related Articles

• Pydantic V2: Data Validation in Python — Learn how to validate data schemas with Pydantic, often used alongside SQLAlchemy models
• Build a REST API with FastAPI in Python — Combine SQLAlchemy ORM with FastAPI for modern API development
• Using the OpenAI API in Python — Integrate AI capabilities into your SQLAlchemy-based applications

Now you have a complete, production-ready reference for SQLAlchemy 2.0 ORM. Use this guide to build, query, and maintain your database layer with confidence.