KAHIBARO
12.10. Queries
Table of Contents
Querying with an ORM
Working with an ORM is all about expressing database queries in your programming language instead of writing raw SQL. In this chapter we focus on how to read data using queries, not how to define models or manage sessions, which are covered in their own chapters.
Examples below use Python and SQLAlchemy 2.x style (the most common choice with FastAPI), but the ideas are the same in other ORMs.
Basic query patterns
The ORM gives you an object oriented way to express the same things you would do with SQL:
| SQL idea | Typical ORM pattern |
|---|---|
SELECT * FROM users | select(User) or session.query(User) |
| WHERE filter | .where(...) or .filter(...) |
| ORDER BY | .order_by(...) |
| LIMIT/OFFSET | .limit(...).offset(...) |
| JOIN | join() / relationship-based loading |
| COUNT, SUM | func.count, func.sum |
Below is a minimal setup we will reuse in examples:
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import declarative_base, relationship
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
email = Column(String, unique=True, nullable=False)
full_name = Column(String)
age = Column(Integer)
posts = relationship("Post", back_populates="author")
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True)
title = Column(String, nullable=False)
body = Column(String)
author_id = Column(Integer, ForeignKey("users.id"))
author = relationship("User", back_populates="posts")
We will assume you already have an active Session object, usually called session.
Selecting all records
Select all rows from a table
In SQL:
SELECT * FROM users;With SQLAlchemy 2.x style:
from sqlalchemy import select
stmt = select(User)
result = session.execute(stmt)
users = result.scalars().all()Important details:
select(User)produces aSELECT * FROM users.session.execute(stmt)sends it to the database..scalars()extracts model objects from the rows..all()loads all of them into a list.
Typical usage:
def get_all_users(session):
return session.execute(select(User)).scalars().all()
If you forget .scalars() you will get Row objects instead of User instances, which can be confusing.
Filtering with conditions
Filtering is the ORM equivalent of WHERE in SQL.
Simple equality filter
SQL:
SELECT * FROM users WHERE email = 'john@example.com';ORM:
stmt = select(User).where(User.email == "john@example.com")
user = session.execute(stmt).scalar_one_or_none()
scalar_one_or_none() returns:
- the single result, or
Noneif there is no row, or- raises if there are multiple rows.
Use this pattern often for queries by unique keys.
Multiple conditions
SQL:
SELECT * FROM users WHERE age >= 18 AND age <= 30;ORM:
from sqlalchemy import and_
stmt = select(User).where(
and_(User.age >= 18, User.age <= 30)
)
users = session.execute(stmt).scalars().all()You can also use Python operators chained, which is usually cleaner:
stmt = select(User).where(
User.age >= 18,
User.age <= 30,
)
SQLAlchemy treats multiple arguments to where() as an implicit AND.
OR conditions
SQL:
SELECT * FROM users WHERE age < 18 OR age > 65;ORM:
from sqlalchemy import or_
stmt = select(User).where(
or_(User.age < 18, User.age > 65)
)
users = session.execute(stmt).scalars().all()IN and NOT IN
SQL:
SELECT * FROM users WHERE id IN (1, 2, 3);ORM:
user_ids = [1, 2, 3]
stmt = select(User).where(User.id.in_(user_ids))
users = session.execute(stmt).scalars().all()For NOT IN:
stmt = select(User).where(~User.id.in_(user_ids))
The ~ is “NOT” here.
LIKE and ILIKE (pattern matching)
SQL:
SELECT * FROM users WHERE full_name LIKE 'John%';ORM:
stmt = select(User).where(User.full_name.like("John%"))
users = session.execute(stmt).scalars().all()Case insensitive pattern matching (PostgreSQL):
stmt = select(User).where(User.full_name.ilike("%smith%"))
Important rule: Never build filters by string concatenation with user input. Always use ORM expressions like User.email == user_email or parameters to avoid SQL injection.
Ordering and limiting results
ORDER BY
SQL:
SELECT * FROM users ORDER BY age DESC, id ASC;ORM:
from sqlalchemy import desc
stmt = select(User).order_by(desc(User.age), User.id)
users = session.execute(stmt).scalars().all()
You can also use User.age.desc():
stmt = select(User).order_by(User.age.desc(), User.id.asc())LIMIT and OFFSET
SQL:
SELECT * FROM users ORDER BY id LIMIT 10 OFFSET 20;ORM:
stmt = (
select(User)
.order_by(User.id)
.limit(10)
.offset(20)
)
users = session.execute(stmt).scalars().all()
This combination of order_by, limit, and offset is the classic pagination pattern for APIs.
Getting single objects
There are several helpers for fetching one row.
`first()`, `one()`, `one_or_none()`, `scalar_one_or_none()`
Common patterns:
# First row or None
user = session.execute(
select(User).where(User.email == email)
).scalars().first()
# Exactly one row, or raise if 0 or > 1
user = session.execute(
select(User).where(User.id == user_id)
).scalar_one()
# One row or None, raise if > 1
user = session.execute(
select(User).where(User.id == user_id)
).scalar_one_or_none()Typical use in an API:
def get_user_or_404(session, user_id: int):
user = session.execute(
select(User).where(User.id == user_id)
).scalar_one_or_none()
if user is None:
# In a FastAPI app, you might raise HTTPException(status_code=404)
raise ValueError("User not found")
return userProjections: selecting specific columns
You do not always need full model instances. You can select specific columns.
SQL:
SELECT id, email FROM users;ORM:
stmt = select(User.id, User.email)
rows = session.execute(stmt).all()
for row in rows:
print(row.id, row.email)
You get Row objects with those fields.
You can also select expressions:
from sqlalchemy import func
stmt = select(
User.id,
User.email,
func.length(User.full_name).label("name_length"),
)
rows = session.execute(stmt).all()
for row in rows:
print(row.id, row.email, row.name_length)
Labeling expressions with .label("...") gives them a column name.
Aggregations and counting
Counting rows
SQL:
SELECT COUNT(*) FROM users;ORM:
from sqlalchemy import func
stmt = select(func.count(User.id))
total_users = session.execute(stmt).scalar_one()Filter and count:
stmt = select(func.count(User.id)).where(User.age >= 18)
adult_count = session.execute(stmt).scalar_one()Grouping and aggregates
SQL:
SELECT age, COUNT(*) FROM users GROUP BY age;ORM:
stmt = (
select(User.age, func.count(User.id).label("count"))
.group_by(User.age)
)
rows = session.execute(stmt).all()
for age, count in rows:
print(age, count)Multiple aggregates:
stmt = select(
func.min(User.age),
func.max(User.age),
func.avg(User.age),
)
min_age, max_age, avg_age = session.execute(stmt).one()
Important pattern: Use func for database functions like COUNT, MAX, LOWER, etc. This keeps your query portable and parameterized.
Joining related tables
Joins let you query across relationships in one go.
Simple join
SQL:
SELECT * FROM posts JOIN users ON posts.author_id = users.id;ORM:
from sqlalchemy import select, join
stmt = (
select(Post, User)
.join(User, Post.author_id == User.id)
)
rows = session.execute(stmt).all()
for post, author in rows:
print(post.title, "by", author.email)
Note that each row from execute contains both models.
Join using relationships
If you have relationships defined, you can join via the relationship.
stmt = select(Post).join(Post.author).where(User.email == "john@example.com")
posts = session.execute(stmt).scalars().all()
Here, Post.author refers to the relationship. SQLAlchemy builds the join condition automatically.
Filtering across relationships
Find all users who have at least one post:
stmt = (
select(User)
.join(User.posts)
.group_by(User.id)
)
users_with_posts = session.execute(stmt).scalars().all()Filter posts by author age:
stmt = (
select(Post)
.join(Post.author)
.where(User.age >= 18)
)
posts = session.execute(stmt).scalars().all()Eager loading vs lazy loading
When you access user.posts, by default the ORM will often do a separate query the moment you access the attribute. This is called lazy loading.
Lazy loading example:
users = session.execute(select(User)).scalars().all()
for user in users:
# This may run another SQL query each time
print(user.email, len(user.posts))This can cause many small queries, sometimes called the N+1 query problem.
Avoiding N+1 with eager loading
You can ask the ORM to load related data in the same query.
from sqlalchemy.orm import selectinload
stmt = select(User).options(selectinload(User.posts))
users = session.execute(stmt).scalars().all()
for user in users:
# Now user.posts is already loaded
print(user.email, len(user.posts))There are several eager loading strategies:
| Strategy | Purpose |
|---|---|
selectinload | Does a separate IN query for related rows, efficient for many parents |
joinedload | Uses JOIN to load relationships in one big query |
Example with joinedload:
from sqlalchemy.orm import joinedload
stmt = select(Post).options(joinedload(Post.author))
posts = session.execute(stmt).scalars().all()
for post in posts:
# post.author already loaded
print(post.title, post.author.email)Important rule: Be explicit about loading strategies when you need related data for many rows. This avoids hidden extra queries and performance issues.
Querying JSON fields (PostgreSQL example)
If your model has a JSON column:
from sqlalchemy import JSON
class Event(Base):
__tablename__ = "events"
id = Column(Integer, primary_key=True)
type = Column(String, nullable=False)
payload = Column(JSON, nullable=False)You can filter on JSON keys:
from sqlalchemy import cast, String
# payload: {"user_id": 123, "source": "mobile"}
stmt = select(Event).where(Event.payload["user_id"].as_integer() == 123)
events = session.execute(stmt).scalars().all()Or check nested structures, depending on your database capabilities. The exact syntax can depend on backend, but the idea is the same: access JSON keys through the column object.
Combining queries into reusable functions
You rarely put query logic directly in your route handlers. You usually wrap it in functions or repository methods.
Example "repository" style:
from typing import Sequence, Optional
from sqlalchemy import select
class UserRepository:
def __init__(self, session):
self.session = session
def list_users(
self,
limit: int = 100,
offset: int = 0,
) -> Sequence[User]:
stmt = (
select(User)
.order_by(User.id)
.limit(limit)
.offset(offset)
)
return self.session.execute(stmt).scalars().all()
def get_by_email(self, email: str) -> Optional[User]:
stmt = select(User).where(User.email == email)
return self.session.execute(stmt).scalar_one_or_none()
def list_by_min_age(self, min_age: int) -> Sequence[User]:
stmt = select(User).where(User.age >= min_age)
return self.session.execute(stmt).scalars().all()Then your API code becomes simple:
def list_users_endpoint(session=Depends(get_session)):
repo = UserRepository(session)
return repo.list_users(limit=50, offset=0)This keeps your queries testable and easier to change.
Pagination, filtering, and searching in APIs
Backend APIs usually combine these query patterns together.
Imagine an endpoint:
GET /users?search=john&min_age=18&max_age=40&limit=20&offset=0You can build the query step by step.
from typing import Optional
from sqlalchemy import select, and_, or_
def search_users(
session,
search: Optional[str] = None,
min_age: Optional[int] = None,
max_age: Optional[int] = None,
limit: int = 20,
offset: int = 0,
):
stmt = select(User)
# Dynamic filters
conditions = []
if search:
like_pattern = f"%{search}%"
conditions.append(
or_(
User.email.ilike(like_pattern),
User.full_name.ilike(like_pattern),
)
)
if min_age is not None:
conditions.append(User.age >= min_age)
if max_age is not None:
conditions.append(User.age <= max_age)
if conditions:
stmt = stmt.where(and_(*conditions))
stmt = stmt.order_by(User.id).limit(limit).offset(offset)
users = session.execute(stmt).scalars().all()
return usersThis pattern lets you assemble flexible queries based on request parameters without ever touching raw SQL.
Common pitfalls and how to avoid them
- Loading too much data
Example of a problem:
users = session.execute(select(User)).scalars().all() # thousands of usersBetter: always limit or paginate.
users = session.execute(
select(User).order_by(User.id).limit(100)
).scalars().all()- N+1 problem
Looping and accessing related objects inside the loop without eager loading:
users = session.execute(select(User)).scalars().all()
for user in users:
# Each `user.posts` might trigger its own query
print(user.email, len(user.posts))
Fix with selectinload:
stmt = select(User).options(selectinload(User.posts))
users = session.execute(stmt).scalars().all()- Session closed or expired
If you access user.posts after the session is closed, lazy loading will fail. If you must use objects after the session is closed, load relationships eagerly.
- Mixing raw SQL and ORM expressions incorrectly
If you need raw SQL, use text() or session.execute(text(...)). Do not concatenate raw strings with ORM expressions. Keep one approach per query.
Summary
In this chapter you learned how to:
- Select all records and single records with
select,execute, and scalar helpers. - Filter using equality, ranges,
IN,LIKE, and boolean combinations. - Order, limit, and offset results for pagination.
- Select specific columns and use aggregates like
COUNTandAVG. - Join related tables and filter across relationships.
- Control loading of relationships to avoid N+1 problems with
selectinloadandjoinedload. - Wrap queries in reusable repository methods.
- Build real world API filters and search features from ORM queries.
These querying techniques are the core of reading data from your database through an ORM. In the next chapters you will see how to combine them with transactions, connection pooling, and the repository pattern to build robust backend applications.
Views: 7