SQLAlchemy 2.0 Tutorial: ORM, Sessions, and select()

A page that lists 50 projects and their tasks took four seconds to load. The database was healthy and every query used an index. The page simply ran 51 queries: one for the projects, then one more per project to fetch its tasks. The ORM had done exactly what the code asked, one innocent attribute access at a time.

SQLAlchemy is the most widely used database toolkit for Python. Its ORM (object-relational mapper) lets you work with rows as Python objects. This SQLAlchemy 2.0 tutorial covers the 2.0 release line, which became final on 26 January 2023 and changes how you declare models and write queries.

You need Python 3.11 and SQLAlchemy 2.0.x. The examples use SQLite, which ships with Python, so there is no database server to install. You should know basic SQL (SELECT, JOIN, WHERE) and how to define a Python class.

My position: use the ORM for the transactional core of your application, and treat SQL visibility as non-negotiable. An ORM you cannot see through will eventually surprise you in production.

Install SQLAlchemy 2.0 and create an engine

python --version
python -m venv .venv
source .venv/bin/activate          # macOS and Linux
.venv\Scripts\Activate.ps1         # Windows PowerShell
python -m pip install "sqlalchemy>=2.0,<2.1"
python -c "import sqlalchemy; print(sqlalchemy.__version__)"

2.0.4

Everything starts with an engine. The engine holds a pool of database connections and knows which SQL dialect to speak.

from sqlalchemy import create_engine

engine = create_engine('sqlite:///tasks.db', echo=True)

The URL names the database. sqlite:///tasks.db is a file in the current directory, and sqlite:// with no path is an in-memory database that disappears when the program ends. The echo=True option prints every SQL statement. Keep it on while you learn, because it shows you what the ORM really does. Create one engine per database for the whole application, not one per request.

Define models with Mapped and mapped_column

A model is a class mapped to a table. In 2.0, you declare each column with a type annotation wrapped in Mapped[...].

from sqlalchemy import ForeignKey, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship


class Base(DeclarativeBase):
    pass


class Project(Base):
    __tablename__ = 'project'

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)
    tasks: Mapped[list['Task']] = relationship(
        back_populates='project', cascade='all, delete-orphan'
    )


class Task(Base):
    __tablename__ = 'task'

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    done: Mapped[bool] = mapped_column(default=False)
    notes: Mapped[str | None]
    project_id: Mapped[int] = mapped_column(ForeignKey('project.id'))
    project: Mapped[Project] = relationship(back_populates='tasks')

The annotation carries real information. Mapped[str] becomes a NOT NULL text column. Mapped[str | None] becomes a nullable one, with no mapped_column() call needed. Mapped[list['Task']] declares a one-to-many relationship, and back_populates links the two sides so that both stay in sync in memory.

These annotations are what make 2.0 different. Type checkers now understand that task.title is a str and that project.tasks is a list of Task, with no plugin. If annotations are new to you, see this guide to type hints and mypy.

The models look like Python dataclasses, and the resemblance is deliberate. SQLAlchemy 2.0 can even generate dataclass-style constructors through its MappedAsDataclass mixin. The plain form above already accepts keyword arguments, such as Task(title='Write docs').

Create the tables from the models:

Base.metadata.create_all(engine)

create_all creates missing tables and never alters existing ones. It suits tutorials and tests. For a real application whose schema changes over time, use Alembic, SQLAlchemy’s migration tool.

Work with a Session

A Session is your workspace for one unit of work. It tracks the objects you load and create, and it writes all pending changes in one transaction when you commit.

from sqlalchemy.orm import Session

with Session(engine) as session:
    project = Project(name='website')
    project.tasks.append(Task(title='Write landing page'))
    session.add(project)
    session.commit()
    print(project.id, project.tasks[0].id)

1 1

Three things happened. session.add() registered the project, and the task came along through the relationship. commit() sent the INSERT statements and ended the transaction. The database assigned the IDs, and SQLAlchemy loaded them back into the objects.

Object states in a Session

  Task(...)            transient    Python object only, unknown to the session
     | session.add()
     v
  pending              waiting for the next flush
     | flush or commit
     v
  persistent           has a row and an identity in this session
     | session closes
     v
  detached             still a Python object, but lazy loading no longer works

Updates need no special call. Change an attribute on a persistent object, and the session notices:

with Session(engine) as session:
    task = session.get(Task, 1)
    task.done = True
    session.commit()          # emits UPDATE task SET done=1 WHERE id=1

This pattern is called unit of work. You modify objects, and the session works out the SQL. If anything raises before commit(), leaving the with block rolls the transaction back. To make that explicit, use with Session(engine) as session, session.begin():, which commits on success and rolls back on error.

Query with select()

In 2.0, you build a statement with select() and hand it to the session. The older session.query() still works, but the documentation now calls it legacy.

from sqlalchemy import select

with Session(engine) as session:
    # ORM objects: use scalars()
    statement = select(Task).where(Task.done.is_(False)).order_by(Task.id)
    open_tasks = session.scalars(statement).all()

    # one object or None
    first = session.scalars(select(Task).where(Task.title == 'Write docs')).first()

    # by primary key
    task = session.get(Task, 1)

    # individual columns across a join: use execute()
    rows = session.execute(
        select(Task.title, Project.name).join(Task.project)
    ).all()

The choice between scalars() and execute() confuses most newcomers. session.execute() returns rows, which behave like tuples. When you select a whole entity, each row is a one-element tuple, such as (Task(...),). session.scalars() unwraps that first element for you. Therefore, use scalars() when you select one entity, and execute() when you select several columns.

The trade-off is verbosity. The 2.0 form takes more characters for simple lookups. In return, ORM and Core queries share one syntax, results are typed, and the same statement objects work with the async API.

Relationships and the N+1 problem

By default, a relationship loads lazily. SQLAlchemy runs a query for project.tasks the first time you touch that attribute. Lazy loading is convenient for one object and expensive in a loop.

# WRONG: one query for the projects, then one more per project
projects = session.scalars(select(Project)).all()
for project in projects:
    print(project.name, len(project.tasks))

# RIGHT: two queries in total, however many projects exist
from sqlalchemy.orm import selectinload

statement = select(Project).options(selectinload(Project.tasks))
projects = session.scalars(statement).all()
for project in projects:
    print(project.name, len(project.tasks))

The wrong version is the N+1 problem: 1 query for the list plus N for the related rows. selectinload fetches all the tasks for all the loaded projects in a second query that uses WHERE project_id IN (...).

The myth: the ORM writes efficient SQL for you

Many teams adopt an ORM believing they no longer need to think about SQL. The ORM writes correct SQL for each statement you give it. It has no idea that your loop is about to trigger the same lookup 50 times. Efficiency depends on how many statements you cause, and only you can see the loop.

Count your queries

Here is the technique that turns N+1 from a production surprise into a failing test. SQLAlchemy fires an event before every statement. A few lines of code can count them.

The complete script below builds the models, seeds data, runs queries, and measures lazy loading against selectinload. Save it as app.py.

from contextlib import contextmanager

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


class Base(DeclarativeBase):
    pass


class Project(Base):
    __tablename__ = 'project'

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100), unique=True)
    tasks: Mapped[list['Task']] = relationship(
        back_populates='project', cascade='all, delete-orphan'
    )


class Task(Base):
    __tablename__ = 'task'

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    done: Mapped[bool] = mapped_column(default=False)
    notes: Mapped[str | None]
    project_id: Mapped[int] = mapped_column(ForeignKey('project.id'))
    project: Mapped[Project] = relationship(back_populates='tasks')

    def __repr__(self) -> str:
        return f'Task(id={self.id}, title={self.title!r}, done={self.done})'


@contextmanager
def count_queries(engine):
    counter = {'count': 0}

    def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
        counter['count'] += 1

    event.listen(engine, 'before_cursor_execute', before_cursor_execute)
    try:
        yield counter
    finally:
        event.remove(engine, 'before_cursor_execute', before_cursor_execute)


def seed(session: Session) -> None:
    for name in ('website', 'mobile'):
        project = Project(name=name)
        project.tasks = [Task(title=f'{name} task {n}') for n in range(1, 4)]
        session.add(project)
    session.commit()


def main() -> None:
    engine = create_engine('sqlite://')
    Base.metadata.create_all(engine)

    with Session(engine) as session:
        seed(session)

        statement = (
            select(Task).where(Task.done.is_(False)).order_by(Task.id).limit(2)
        )
        print(session.scalars(statement).all())

        task = session.get(Task, 1)
        task.done = True
        session.commit()

        rows = session.execute(
            select(Task.title, Project.name)
            .join(Task.project)
            .where(Task.done.is_(True))
        ).all()
        print(rows)

    with Session(engine) as session, count_queries(engine) as counter:
        for project in session.scalars(select(Project)).all():
            len(project.tasks)
        print(f"lazy loading: {counter['count']} queries")

    with Session(engine) as session, count_queries(engine) as counter:
        statement = select(Project).options(selectinload(Project.tasks))
        for project in session.scalars(statement).all():
            len(project.tasks)
        print(f"selectinload: {counter['count']} queries")


if __name__ == '__main__':
    main()

python app.py

[Task(id=1, title='website task 1', done=False), Task(id=2, title='website task 2', done=False)]
[('website task 1', 'website')]
lazy loading: 3 queries
selectinload: 2 queries

With two projects, lazy loading runs 3 queries and selectinload runs 2. The gap looks trivial here. With 500 projects, it is 501 against 2. My rule: every endpoint that returns a list gets a test that asserts its query count. Wrap count_queries in a pytest fixture, and an accidental N+1 fails the build instead of slowing production.

Transactions, expiry, and detached objects

Two session behaviors cause most of the confusing errors.

First, commit() expires every loaded object. The next attribute access reloads the row, so that you never read stale data after a transaction ends. That refresh is one more query per object. You can disable it with Session(engine, expire_on_commit=False) when you need to use objects after committing, which is common in web handlers that build a response.

Second, an object that outlives its session is detached. Touching an unloaded relationship on it fails:

sqlalchemy.orm.exc.DetachedInstanceError: Parent instance <Project at 0x7f...> is not bound
to a Session; lazy load operation of attribute 'tasks' cannot proceed

The fix is not a longer-lived session. Load what you need while the session is open, with selectinload or joinedload, or convert the objects to plain data before the session closes.

How real systems structure SQLAlchemy code

• One engine, one session factory. The application creates the engine at startup and a sessionmaker(engine) next to it. Code asks the factory for sessions and never builds engines on the fly.
• One session per request or job. A web request opens a session, does its work, commits or rolls back, and closes it. Sessions are never shared between threads or stored in globals.
• Migrations through Alembic. Schema changes live in versioned migration scripts that run during deployment. create_all() stays in tests.
• Explicit loading in list endpoints. Queries that feed lists declare their loader options. Some teams set lazy='raise' on relationships so that any unplanned lazy load fails loudly.
• Core for bulk work. Large imports and reports use Core statements or raw SQL through the same engine, and skip the per-object overhead of the ORM.

A mistake I have seen in production is a module-level session = Session(engine) shared by every request in a threaded web server. Under load, two requests interleaved their changes in the same transaction, and one user’s rollback discarded another user’s update. The symptom was rare, unreproducible data loss. Moving to one session per request, created by a sessionmaker, removed it.

Choosing between the ORM, Core, and raw SQL: a decision framework

1. Are you creating, updating, and reading individual business objects? Use the ORM. Unit of work and relationships save real effort here.
2. Are you returning a list with related data? Use the ORM with explicit loader options, and assert the query count in a test.
3. Are you inserting or updating many thousands of rows? Use Core insert() and update() statements, which skip object tracking.
4. Is it a complex report with window functions or database-specific features? Write SQL with text() and run it through the same engine.
5. Is it a one-file script with two queries? The standard library’s sqlite3 module, or a plain driver, may be all you need.

When NOT to use the ORM

• Bulk data movement. Loading a million rows as ORM objects spends most of its time building and tracking Python objects. Core or the database’s own bulk loader is far faster.
• Analytical queries. Aggregations across many tables are clearer in SQL. Forcing them through ORM constructs produces code that nobody can read or tune.
• Tiny scripts. Models, sessions, and an engine are overhead for a script that runs one SELECT.

Common mistakes

• Lazy loading inside a loop. Each iteration runs a query. Page load time grows in step with the number of rows.
• Sharing a session across threads or requests. Sessions are not thread-safe. Changes from different requests mix in one transaction.
• Using scalars() for multi-column selects. scalars() returns only the first column of each row, so the other columns vanish silently.
• Comparing booleans and None with Python operators. Task.done is False is evaluated by Python immediately and is always False. Use Task.done.is_(False) and Task.notes.is_(None).
• Relying on create_all for schema changes. It never alters an existing table. A new column in the model does not appear in the database, and queries fail.
• Leaving echo=True on in production. Every statement and its parameters go to the logs, which costs performance and can expose personal data.

Key takeaways

• Declare models with DeclarativeBase, Mapped[...], and mapped_column().
• Create one engine per database, and one short-lived session per unit of work.
• Build queries with select(), then use scalars() for entities and execute() for columns.
• Lazy loading is the default, so add selectinload() or joinedload() wherever you loop.
• Count queries with the before_cursor_execute event, and assert the count in tests.
• Use Alembic for schema changes, and keep create_all() for tests.
• Drop to Core or raw SQL for bulk writes and reports.

FAQ

What is new in SQLAlchemy 2.0?

SQLAlchemy 2.0 adds typed models with Mapped and mapped_column(), makes select() the standard query style for both ORM and Core, and treats the older session.query() API as legacy.

Is session.query() removed in SQLAlchemy 2.0?

No. session.query() still works in 2.0, but the documentation marks it as legacy. New code should use select() with session.execute() or session.scalars().

What is the difference between session.execute() and session.scalars()?

session.execute() returns rows, which behave like tuples. session.scalars() returns the first element of each row, which is what you want when you select a single ORM entity.

What is the N+1 problem in SQLAlchemy?

It happens when code loads a list with one query and then triggers one more query per item to load a relationship. Eager loading with selectinload() or joinedload() reduces it to one or two queries.

Should I use SQLAlchemy ORM or Core?

Use the ORM for transactional work on individual objects, where relationships and change tracking help. Use Core for bulk operations and complex queries where you want direct control over the SQL.

Know every query your code sends

SQLAlchemy 2.0 gives you typed models and one consistent way to query. The habits that keep it fast are older than the release: short sessions, deliberate loading, and attention to the SQL that actually runs. Turn on echo while you develop, and count queries in your tests.

Rule of thumb: if you cannot say how many queries an endpoint runs, it runs too many.