Querying with SQLAlchemy

Reviewed & published by Brayan K

The SQLAlchemy query API lets you build database reads in Python — selecting, filtering, sorting, joining, and aggregating rows — and have them translated into efficient SQL for you.

Part of the free Flask course at LearnCodingFast — hands-on lessons with worked examples and the output they print, plus practice exercises and a quick quiz.

In this lesson you'll use select() with where, order_by, limit, joins across tables, and aggregate functions like func.count — all against a seeded in-memory database.

What You'll Learn in This Lesson

• Building reads with select() and scalars()
• Filtering rows with where()
• Sorting with order_by (asc and desc)
• Slicing results with limit() and offset()
• Joining across related tables
• Aggregating with func.count and group_by

1 select, where, order_by, first & all

Every read starts with select(Model). You refine it with .where(...) to filter, .order_by(...) to sort, and then execute it. Wrapping in session.scalars(stmt) returns the model objects; .all() gives a list and .first() gives the first match or None.

The runnable example seeds three books, then sorts by year, filters to recent titles, and fetches a single first result.

# queries.py — select, where, order_by (runnable, seeded sqlite)
from sqlalchemy import create_engine, String, Integer, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase): pass

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

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

with Session(engine) as s:
    s.add_all([
        Book(title="1984", year=1949),
        Book(title="Animal Farm", year=1945),
        Book(title="Brave New World", year=1932),
    ])
    s.commit()

    # order_by, ascending
    by_year = s.scalars(select(Book).order_by(Book.year)).all()
    print("oldest first:", [b.title for b in by_year])

    # where: only books after 1940
    recent = s.scalars(select(Book).where(Book.year > 1940)).all()
    print("after 1940:", [b.title for b in recent])

    # first(): one result or None
    print("first by title:", s.scalars(select(Book).order_by(Book.title)).first().title)

# ✅ Expected output:
# oldest first: ['Brave New World', 'Animal Farm', '1984']
# after 1940: ['1984', 'Animal Farm']
# first by title: 1984

Output: oldest-first list, then ['1984', 'Animal Farm'] for post-1940 titles, then 1984 as the alphabetically first title.

2 Joining Across Tables

When your data spans two tables — books and their authors — a join lets you filter or select using columns from both. You call .join(OtherModel) on the statement and SQLAlchemy figures out the ON clause from the foreign key.

The example finds every book written by a specific author by joining Book to Author and filtering on the author's name.

# queries.py — joins across related tables (runnable)
from sqlalchemy import create_engine, ForeignKey, String, Integer, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase): pass

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    books: Mapped[list["Book"]] = relationship(back_populates="author")

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

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

with Session(engine) as s:
    orwell = Author(name="Orwell")
    huxley = Author(name="Huxley")
    s.add_all([
        orwell, huxley,
        Book(title="1984", year=1949, author=orwell),
        Book(title="Animal Farm", year=1945, author=orwell),
        Book(title="Brave New World", year=1932, author=huxley),
    ])
    s.commit()

    # join Book -> Author, filter on the author's name, select titles
    stmt = select(Book.title).join(Author).where(Author.name == "Orwell")
    print("Orwell's books:", s.scalars(stmt).all())

# ✅ Expected output:
# Orwell's books: ['1984', 'Animal Farm']

Output: Orwell's books: ['1984', 'Animal Farm'] — selected by joining the two tables and filtering on the author.

3 Aggregates, Grouping & limit/offset

To compute summaries, use functions from sqlalchemy.func. func.count() counts rows; combined with .group_by(...) you get counts per group. Because aggregates return values (not entities), read them with session.execute(...).all() for the full rows, or session.scalar(...) for a single number.

.limit(n) and .offset(m) slice results — the foundation of pagination, covered next.

# queries.py — aggregates, group_by, limit/offset (runnable)
from sqlalchemy import create_engine, ForeignKey, String, Integer, select, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session

class Base(DeclarativeBase): pass

class Author(Base):
    __tablename__ = "authors"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))

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"))

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

with Session(engine) as s:
    orwell = Author(name="Orwell"); huxley = Author(name="Huxley")
    s.add_all([orwell, huxley])
    s.flush()   # assigns ids so we can reference them
    s.add_all([
        Book(title="1984", author_id=orwell.id),
        Book(title="Animal Farm", author_id=orwell.id),
        Book(title="Brave New World", author_id=huxley.id),
    ])
    s.commit()

    # total number of books (one scalar value)
    print("total books:", s.scalar(select(func.count()).select_from(Book)))

    # count of books per author (group_by) -> list of (name, count) rows
    stmt = (select(Author.name, func.count(Book.id))
            .join(Book, Book.author_id == Author.id)
            .group_by(Author.name).order_by(Author.name))
    print("per author:", s.execute(stmt).all())

    # limit + offset: page 1 is the first 2 titles
    page = s.scalars(select(Book).order_by(Book.id).limit(2).offset(0)).all()
    print("first two:", [b.title for b in page])

# ✅ Expected output:
# total books: 3
# per author: [('Huxley', 1), ('Orwell', 2)]
# first two: ['1984', 'Animal Farm']

🎯 Your Turn: Fill in the Blanks

Complete the queries. Replace each ___ to filter, sort descending, and count.

# 🎯 YOUR TURN: replace each ___ to finish the queries
from sqlalchemy import create_engine, String, Integer, select, func, desc
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase): pass

class Product(Base):
    __tablename__ = "products"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))
    price: Mapped[int] = mapped_column(Integer)

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

with Session(engine) as s:
    s.add_all([Product(name="Pen", price=2), Product(name="Mug", price=9), Product(name="Book", price=15)])
    s.commit()

    # only products priced over 5
    pricey = s.scalars(select(Product).___(Product.price > 5)).all()   # filter method
    print("over 5:", [p.name for p in pricey])

    # most expensive first
    sorted_desc = s.scalars(select(Product).order_by(___(Product.price))).all()  # descending
    print("priciest first:", [p.name for p in sorted_desc])

    # how many products in total?
    print("count:", s.scalar(select(func.___()).select_from(Product)))  # aggregate

Common Errors & Quick Reference

❌ Got a Row tuple, expected an object

You used execute() when you wanted whole entities. For select(User) use scalars() to get User objects; use execute() only when selecting multiple columns or aggregates.

❌ AttributeError: 'NoneType' has no attribute …

first() returned None because nothing matched. Check the result before reading attributes, or use one_or_none() intentionally.

❌ Comparing with == None or using Python `and`

In a where(), use Column.is_(None) instead of == None, and combine conditions with and_()/or_(), not Python's and/or.

🏆 Mini Challenge: Top Spenders

Find which customer placed the most orders.

• Join orders to customers
• Group by customer name and count orders
• Sort by the count, highest first, and take the top row

# 🏆 Mini Challenge: most orders per customer
from sqlalchemy import create_engine, ForeignKey, String, select, func, desc
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase): pass

class Customer(Base):
    __tablename__ = "customers"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(50))

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"))

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

with Session(engine) as s:
    ada = Customer(name="Ada"); bob = Customer(name="Bob")
    s.add_all([ada, bob]); s.flush()
    s.add_all([
        Order(customer_id=ada.id), Order(customer_id=ada.id), Order(customer_id=ada.id),
        Order(customer_id=bob.id),
    ])
    s.commit()

    stmt = (select(Customer.name, func.count(Order.id).label("n"))
            .join(Order, Order.customer_id == Customer.id)
            .group_by(Customer.name)
            .order_by(desc("n")))
    rows = s.execute(stmt).all()
    print("ranking:", rows)
    print("top spender:", rows[0][0])

❓ Frequently Asked Questions

Lesson complete — you can ask your database anything!

You can build reads with select(), filter with where, sort with order_by, join related tables, and summarize with func.count and group_by.

Practice quiz

Continue this course

• Previous: SQLAlchemy Relationships
• Next: Pagination — Split long result sets into pages with offset/limit and page links
• Quick reference: Flask cheat sheet

Frequently asked questions

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

session.scalars(stmt) returns the first column of each row as plain objects — perfect when you select whole entities like select(User), giving you User objects directly. session.execute(stmt) returns full Row tuples, which you want when you select multiple columns or aggregates like select(User.name, func.count(...)).

How do I filter rows?

Use .where() on a select statement with a comparison built from the model's columns, for example select(Book).where(Book.year > 1940). You can chain multiple conditions, or combine them with and_() / or_(). In the older Query API the equivalent is Book.query.filter(Book.year > 1940).

What is the difference between first() and all()?

all() returns a list of every matching result, even if empty. first() returns just the first result or None if there are none, which is handy for lookups by a unique value. There is also one(), which returns exactly one row and raises if there are zero or more than one.

How do I count or sum without loading every row?

Use SQL aggregate functions from sqlalchemy.func, such as func.count() or func.sum(), inside a select. select(func.count()).select_from(Book) returns just the number, computed by the database, instead of pulling all rows into Python.