SQLAlchemy from Scratch: Hands-On Tutorial:

Introduction to SQLAlchemy

SQLAlchemy is the most widely used SQL toolkit and Object-Relational Mapping (ORM) library for Python. Since its initial release in 2006, it has become the de facto standard for database interaction in Python applications, powering everything from small scripts to large-scale production systems at companies like Reddit, Yelp, and Dropbox. Whether you are building a simple CRUD application or a complex data-driven service, SQLAlchemy provides a flexible and powerful abstraction over raw SQL.

At its core, SQLAlchemy offers two distinct modes of operation: the Core, which is a schema-centric SQL expression language, and the ORM, which maps Python classes to database tables. This tutorial will walk you through both, with hands-on examples you can run immediately.

Why SQLAlchemy Matters

Writing raw SQL in Python applications quickly becomes painful. You have to manage connections, escape values to prevent SQL injection, handle different dialects across database engines, and manually convert rows into Python objects. SQLAlchemy solves all of these problems and more:

• Database Agnosticism: Write your code once and switch between PostgreSQL, MySQL, SQLite, Oracle, and others with minimal changes.
• SQL Injection Protection: Parameter binding is built into every query, keeping your application secure by default.
• Pythonic Data Access: Interact with rows as Python objects with attributes, methods, and relationships.
• Schema Management: Define your schema in Python and let SQLAlchemy generate the DDL for you.
• Eager and Lazy Loading: Fine-tune how related data is loaded to optimize performance.
• Mature Ecosystem: Integrates seamlessly with Flask, FastAPI, Alembic (for migrations), and pandas.

Installation and Setup

SQLAlchemy 2.0 introduced a modern, type-annotated API that we will use throughout this tutorial. Install it with pip:

pip install sqlalchemy

For this tutorial, we will use SQLite, which is built into Python and requires no separate server. Everything you learn here applies equally to PostgreSQL or MySQL — you only change the connection string.

Understanding the Architecture

Before writing code, it helps to understand the two layers of SQLAlchemy:

The Core (SQL Expression Language)

The Core provides a Pythonic way to construct SQL statements. You work with Table objects, select() constructs, and Connection objects. This is ideal when you want fine-grained control over the SQL being generated or when you are working with existing databases.

The ORM

The ORM builds on top of the Core. You define Python classes decorated with @orm.mapped_as_dataclass or using declarative base classes, and SQLAlchemy maps them to tables. You then interact with a Session object that tracks changes and synchronizes them to the database. The ORM is ideal for domain-driven applications where you think in terms of business objects.

Connecting to a Database

Everything in SQLAlchemy starts with an Engine. The engine is the starting point for any SQLAlchemy application — it holds the connection pool and the database dialect. Create one using the create_engine() function:

from sqlalchemy import create_engine

# SQLite in-memory database (great for testing)
engine = create_engine("sqlite:///:memory:", echo=True)

# For a file-based SQLite database:
# engine = create_engine("sqlite:///app.db")

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

# For MySQL:
# engine = create_engine("mysql+pymysql://user:password@localhost/mydb")

The echo=True parameter tells SQLAlchemy to log every SQL statement it executes, which is invaluable for learning and debugging. In production, you would typically set this to False and configure proper logging instead.

Defining Models with the ORM

In SQLAlchemy 2.0, the recommended way to define models is using DeclarativeBase with type annotations. Let us build a small blog application with authors and posts.

from datetime import datetime
from typing import List, Optional

from sqlalchemy import String, Text, ForeignKey, DateTime, func
from sqlalchemy.orm import DeclarativeBase, 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))
    email: Mapped[str] = mapped_column(String(255), unique=True)

    # A one-to-many relationship: one author has many posts
    posts: Mapped[List["Post"]] = relationship(
        back_populates="author", cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"<Author id={self.id} name={self.name!r}>"


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    body: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now()
    )
    author_id: Mapped[Optional[int]] = mapped_column(ForeignKey("authors.id"))

    # The other side of the relationship
    author: Mapped[Optional["Author"]] = relationship(back_populates="posts")

    def __repr__(self) -> str:
        return f"<Post id={self.id} title={self.title!r}>"

Let us break down what is happening here:

• DeclarativeBase is the base class for all your models. It maintains a registry of all mapped classes.
• Mapped[T] is a type annotation that tells SQLAlchemy the Python type of the column. Optional[int] means the column is nullable.
• mapped_column() configures the column: its type, constraints, and database-level options.
• relationship() defines how objects relate to each other in Python. It does not create a column in the database; it uses the foreign key to load related objects.
• back_populates links both sides of the relationship so that setting post.author automatically updates author.posts and vice versa.

Creating the Schema

Once your models are defined, you can create the corresponding tables in the database with a single call:

Base.metadata.create_all(engine)

This inspects all subclasses of Base and emits the appropriate CREATE TABLE statements. For production applications, you should use Alembic, SQLAlchemy's migration tool, instead of create_all so that you can evolve your schema over time without losing data.

Working with Sessions

The Session is the ORM's interface to the database. It tracks changes to objects and flushes them to the database in a transactional manner. The recommended pattern is to use a sessionmaker factory:

from sqlalchemy.orm import sessionmaker, Session

SessionLocal = sessionmaker(bind=engine)

# Use it as a context manager (SQLAlchemy 2.0 style)
with Session(engine) as session:
    # your database operations here
    session.commit()

For web frameworks like FastAPI or Flask, you typically create a dependency that yields a session per request:

from contextlib import contextmanager

@contextmanager
def get_session():
    session = SessionLocal()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

Creating Records

Let us create some authors and posts. SQLAlchemy makes it easy to build related objects together:

with Session(engine) as session:
    # Create an author with two posts
    alice = Author(
        name="Alice Johnson",
        email="alice@example.com",
        posts=[
            Post(title="Getting Started with SQLAlchemy", body="SQLAlchemy is..."),
            Post(title="Advanced Queries", body="Let us dive deeper..."),
        ],
    )

    bob = Author(
        name="Bob Smith",
        email="bob@example.com",
        posts=[
            Post(title="My First Post", body="Hello world!"),
        ],
    )

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

    print(f"Alice's id: {alice.id}")
    print(f"Bob's id: {bob.id}")

Notice that we never set author_id manually. SQLAlchemy handles it automatically when we commit, because the posts relationship cascades the foreign key assignment.

Querying Data

SQLAlchemy 2.0 introduced the select() statement as the primary way to build queries. This is a significant departure from the older session.query() API and is more composable and type-safe.

Basic Queries

from sqlalchemy import select

with Session(engine) as session:
    # Select all authors
    stmt = select(Author)
    authors = session.execute(stmt).scalars().all()
    for author in authors:
        print(author)

    # Select a single author by primary key
    alice = session.get(Author, 1)
    print(alice)

    # Filter by a column
    stmt = select(Author).where(Author.name == "Bob Smith")
    bob = session.execute(stmt).scalar_one()
    print(bob)

The scalars() method extracts the first column from each row, which is what you want when selecting entire entities. The scalar_one() method returns exactly one result and raises an error if there are zero or multiple matches — useful for enforcing uniqueness assumptions.

Filtering and Ordering

with Session(engine) as session:
    # Authors whose name contains "li"
    stmt = select(Author).where(Author.name.like("%li%"))
    results = session.execute(stmt).scalars().all()

    # Posts ordered by creation date, newest first
    stmt = (
        select(Post)
        .order_by(Post.created_at.desc())
        .limit(10)
    )
    recent_posts = session.execute(stmt).scalars().all()

    # Combine multiple conditions with AND / OR
    from sqlalchemy import or_, and_

    stmt = select(Post).where(
        or_(
            Post.title.like("%SQLAlchemy%"),
            Post.title.like("%Advanced%"),
        )
    )
    matching = session.execute(stmt).scalars().all()

Joins and Relationship Loading

One of the most powerful features of the ORM is automatic relationship loading. When you access author.posts, SQLAlchemy can load the posts lazily on demand or eagerly in a single query.

from sqlalchemy.orm import selectinload, joinedload

with Session(engine) as session:
    # Eager load posts with a separate IN query (recommended for collections)
    stmt = select(Author).options(selectinload(Author.posts))
    authors = session.execute(stmt).scalars().all()
    for author in authors:
        print(f"{author.name} has {len(author.posts)} posts")

    # Join query: posts with their authors
    stmt = (
        select(Post)
        .join(Author)
        .where(Author.name == "Alice Johnson")
    )
    alice_posts = session.execute(stmt).scalars().all()
    for post in alice_posts:
        print(f"  - {post.title}")

    # Aggregate query: count posts per author
    from sqlalchemy import func

    stmt = (
        select(Author.name, func.count(Post.id).label("post_count"))
        .join(Post, isouter=True)
        .group_by(Author.id)
        .order_by(func.count(Post.id).desc())
    )
    for row in session.execute(stmt):
        print(f"{row.name}: {row.post_count} posts")

The selectinload strategy is generally the best choice for loading collections because it avoids the N+1 query problem without producing Cartesian products. For many-to-one relationships, joinedload is often preferable since it uses a single JOIN.

Updating Records

Updating is straightforward: load an object, modify its attributes, and commit. The session automatically tracks which objects have changed.

with Session(engine) as session:
    alice = session.get(Author, 1)
    alice.email = "alice.new@example.com"
    session.commit()
    print(f"Updated email: {alice.email}")

# Bulk update with a WHERE clause
from sqlalchemy import update

with Session(engine) as session:
    session.execute(
        update(Post)
        .where(Post.title.like("%Advanced%"))
        .values(title="Advanced SQLAlchemy Queries")
    )
    session.commit()

Deleting Records

with Session(engine) as session:
    bob = session.get(Author, 2)
    session.delete(bob)  # Also deletes Bob's posts due to cascade
    session.commit()

# Bulk delete
from sqlalchemy import delete

with Session(engine) as session:
    session.execute(
        delete(Post).where(Post.title.like("%Hello%"))
    )
    session.commit()

Using the Core Directly

Sometimes you want the power of SQL without the ORM overhead. The Core lets you work with tables and connections directly:

from sqlalchemy import Table, Column, Integer, String, MetaData

metadata = MetaData()

users = Table(
    "users",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("username", String(50), unique=True),
    Column("email", String(255)),
)

metadata.create_all(engine)

with engine.connect() as conn:
    # Insert
    conn.execute(
        users.insert(),
        [
            {"username": "charlie", "email": "charlie@example.com"},
            {"username": "diana", "email": "diana@example.com"},
        ],
    )
    conn.commit()

    # Select
    result = conn.execute(
        select(users.c.username, users.c.email).where(
            users.c.username == "charlie"
        )
    )
    for row in result:
        print(row)

The Core is useful for ETL pipelines, reporting queries, or any scenario where you want maximum control and performance without object overhead.

Transactions and Error Handling

SQLAlchemy sessions are transactional by default. Every operation is part of a transaction that must be committed or rolled back. Proper error handling is essential:

from sqlalchemy.exc import IntegrityError

with Session(engine) as session:
    try:
        duplicate = Author(name="Alice Clone", email="alice@example.com")
        session.add(duplicate)
        session.commit()
    except IntegrityError:
        session.rollback()
        print("Caught duplicate email — rolled back the transaction.")

Always wrap your session usage in try/except blocks and call rollback() on errors. Failing to do so leaves the session in a broken state where subsequent operations will raise errors.

Best Practices

1. Use the Session as a Context Manager

Always use with Session(engine) as session: or a dependency-injection pattern. This ensures the session is properly closed even if an exception occurs.

2. Avoid the N+1 Query Problem

If you loop over a list of objects and access a relationship on each one, SQLAlchemy will issue a separate query for each access. Use selectinload or joinedload to eager-load relationships:

# Bad: N+1 queries
authors = session.execute(select(Author)).scalars().all()
for a in authors:
    print(len(a.posts))  # One query per author!

# Good: 2 queries total
authors = (
    session.execute(
        select(Author).options(selectinload(Author.posts))
    )
    .scalars()
    .all()
)
for a in authors:
    print(len(a.posts))

3. Use Alembic for Migrations

Never use create_all in production. Use Alembic to generate and apply migration scripts so your schema changes are versioned and reversible.

4. Keep Sessions Short-Lived

Sessions are not thread-safe and hold database connections. Create a new session per unit of work (typically per web request) and close it promptly.

5. Prefer select() Over Query

The legacy session.query() API still works but is deprecated in favor of the select() construct, which is more composable, supports better type checking, and works seamlessly with both Core and ORM.

6. Use Connection Pooling Wisely

SQLAlchemy uses a connection pool by default. For production, tune the pool size and enable pre-ping to detect stale connections:

engine = create_engine(
    "postgresql+psycopg2://user:pass@host/db",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    pool_recycle=3600,
)

7. Validate Your Data

SQLAlchemy does not validate data beyond database constraints. For application-level validation, consider integrating with Pydantic or Marshmallow, or use SQLAlchemy's hybrid properties and validators.

Putting It All Together

Here is a complete, runnable script that ties together everything we have covered:

from datetime import datetime
from typing import List, Optional

from sqlalchemy import String, Text, ForeignKey, DateTime, func, select, create_engine
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    mapped_column,
    relationship,
    Session,
    selectinload,
)


class Base(DeclarativeBase):
    pass


class Author(Base):
    __tablename__ = "authors"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), 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))
    body: Mapped[str] = mapped_column(Text)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now()
    )
    author_id: Mapped[Optional[int]] = mapped_column(ForeignKey("authors.id"))
    author: Mapped[Optional["Author"]] = relationship(back_populates="posts")


def main():
    engine = create_engine("sqlite:///:memory:", echo=False)
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        # Create
        alice = Author(
            name="Alice",
            email="alice@example.com",
            posts=[
                Post(title="Post One", body="Body one"),
                Post(title="Post Two", body="Body two"),
            ],
        )
        bob = Author(
            name="Bob",
            email="bob@example.com",
            posts=[Post(title="Bob's Post", body="Hello")],
        )
        session.add_all([alice, bob])
        session.commit()

        # Read with eager loading
        stmt = select(Author).options(selectinload(Author.posts))
        for author in session.execute(stmt).scalars():
            print(f"{author.name}: {len(author.posts)} posts")

        # Update
        alice.name = "Alice Updated"
        session.commit()

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

        # Verify
        remaining = session.execute(select(Author)).scalars().all()
        print(f"Remaining authors: {[a.name for a in remaining]}")


if __name__ == "__main__":
    main()

Conclusion

SQLAlchemy is a mature, feature-rich library that scales from simple scripts to enterprise applications. By understanding the distinction between Core and ORM, mastering the select() statement, and following best practices around session management and eager loading, you can build robust data layers that are both performant and maintainable. Start with the basics covered here, integrate Alembic for migrations as your schema evolves, and gradually explore advanced features like hybrid properties, polymorphic mappings, and event listeners as your application's needs grow. The investment in learning SQLAlchemy pays dividends across virtually every Python project that touches a relational database.