sqlalchemy-if 0.1.0

Conditional, chainable helpers for SQLAlchemy select statements (Core & ORM) — where_if / order_if / limit_if.

sqlalchemy-if

Conditional, chainable helpers for SQLAlchemy select() statements — works with both Core tables and ORM models (select() is unified in SQLAlchemy 2.0). Kill the if param: ladder in your search/list endpoints.

Installs as sqlalchemy-if, imports as sqla_if — the short name keeps it distinguishable from sqlalchemy itself at the top of your file.

Not an official SQLAlchemy project. sqlalchemy-if is an independent, third-party library that builds on SQLAlchemy; it is not affiliated with, endorsed by, or maintained by the SQLAlchemy project. "SQLAlchemy" is used only to say what this extends. All trademarks belong to their respective owners.

from sqla_if import select_if

stmt = (
    select_if(User)
    .where_if(name, User.name == name)     # applied only if name is not None
    .where_if(min_age, lambda v: User.age >= v)
    .order_if(sort, columns=User)          # "-created_at,name" -> ORDER BY, whitelisted
    .limit_if(limit)                        # 0 means LIMIT 0, not "skip"
    .paginate(page, size)
).build()                                    # -> a plain SQLAlchemy Select

users = session.execute(stmt).scalars().all()

Instead of:

stmt = select(User)
if name is not None:
    stmt = stmt.where(User.name == name)
if min_age is not None:
    stmt = stmt.where(User.age >= min_age)
if sort:
    ...  # parse, validate, order_by
if limit is not None:
    stmt = stmt.limit(limit)
...

Why not just use if?

Because the interesting bugs live in the conditions:

• limit_if(0) should mean LIMIT 0, not "no limit". A truthiness check (if limit:) silently drops it. sqlalchemy-if defaults to is not None, so falsy values that are still valid are honored.
• order_if("password; DROP ...") should be rejected, not passed to getattr. Sort columns are resolved against a whitelist and raise ValueError on anything unknown.

Every helper here exists for a bug like that, not to wrap SQLAlchemy.

Install

pip install sqlalchemy-if      # requires SQLAlchemy >= 1.4 (tested on 1.4 and 2.0)

---

Quick start

Three things and you can write the endpoint. Everything after this is reference — read it when you hit the thing it describes.

1. Build, then execute

Every method returns a new builder (immutable, like SQLAlchemy's own API). .build() hands back a plain Select:

builder = select_if(User).where_if(name, User.name)
stmt = builder.build()                        # a plain SQLAlchemy Select
users = session.execute(stmt).scalars().all()

.build() is required before executing — session.execute() wants a real statement, and the builder isn't one. Only sql_str / print_sql take a builder directly.

The library never touches your session. It composes a statement; running it stays yours.

2. when= — when does a clause apply?

The one concept worth reading up front. Every *_if method takes when=, a value -> bool predicate deciding whether the value counts as "present":

.where_if(name, User.name)                 # applied unless name is None

The default is NOT_NONE (value is not None) — deliberately not truthiness, because 0, False and "" are often valid filter values. limit_if(0) means LIMIT 0, not "no limit".

.limit_if(0)                      # LIMIT 0        <- honored
.limit_if(None)                   # no LIMIT       <- skipped
.where_if("", User.name)          # name = ''      <- honored ("" is a value)
.where_if("", User.name, when=NOT_EMPTY)   # skipped, if that's what you meant

That's 90% of it. There are four more predicates and a tri-state sentinel when you need them — see the when= predicate in full.

3. Plain SQLAlchemy still works

The builder is a superset of Select: anything it doesn't define itself (where, order_by, join, group_by, having, options, ...) is forwarded to the wrapped statement and re-wrapped, so unconditional and conditional clauses chain together freely:

stmt = (
    select_if(User)
    .where(User.is_active)             # always (plain SQLAlchemy)
    .where_if(name, User.name == name) # only if name is not None
    .join(Org)                          # always
    .order_if(sort, columns=User)      # conditional
).build()

There's no walled garden to escape from — and .apply(fn) takes any Select -> Select for the rest.

Typing: forwarding happens at runtime via __getattr__, but builder.pyi declares the common generative methods (where, order_by, join, limit, options, ...) with a Self return type, so chains stay type-checked and autocompleted. Less common methods fall back to Any. The package ships py.typed.

For a whole endpoint in context, see FastAPI.

Your own query methods

Query fragments worth naming — "active users", "visible to this tenant" — can live on a SelectBuilder subclass and chain alongside the helpers. Pass it to select_if with builder_cls=:

from sqla_if import SelectBuilder, select_if

class UserQuery(SelectBuilder):
    __slots__ = ()

    def active(self):
        return self.where(User.is_active.is_(True))

    def in_org(self, org):
        return self.join_if(org, Org).where_if(org, Org.name)

class User(Base):
    ...
    @classmethod
    def query(cls):
        return select_if(cls, builder_cls=UserQuery)

stmt = (
    User.query()
    .active()                    # yours
    .where_if(name, User.name)   # the library's
    .in_org(org)                 # yours
    .paginate(page, size)
).build()

Your methods build on the helpers (in_org above gets join_if's join-once behavior for free), and the chain keeps its class the whole way — including through delegated Select methods — so Self in the stub means your editor follows it too.

Two rules: keep the (stmt, columns) constructor signature (the builder reconstructs itself with it on every call, so a subclass can't add state of its own), and declare __slots__ = () unless you want a __dict__.

Prefer a @classmethod on the model over a mixin from us: the library never imports your models, and never touches your session.

---

Reference

method	applies when	notes
select_if(*entities, columns=None, builder_cls=SelectBuilder)	—	starts the chain; builder_cls for your own query methods
where_if(value, condition, when=NOT_NONE)	when(value)	condition is a bare column (scalar → ==, list → in_), a clause, or lambda v: clause
exclude_if(value, condition, when=NOT_NONE)	when(value)	inverse of where_if — NOT (...): scalar → !=, list → not_in, None → IS NOT NULL
search_if(value, *columns, match=CONTAINS, when=NOT_NONE)	when(value)	LIKE across columns, OR-ed; the term is always escaped
range_if(lo, hi, column, when=NOT_NONE)	per bound	inclusive >= / <=; lo > hi raises
join_if(value, target, onclause=None, isouter=False, when=NOT_NONE)	when(value)	joins once, however many callers ask
order_if(spec, columns=None, tiebreak=None, when=NOT_NONE)	when(spec)	spec: "-created_at,name"; tiebreak for stable paging
order_by_map(key, mapping, default=None, tiebreak=None, when=NOT_NONE)	when(key)	dispatch ORDER BY by key → expression(s); unknown key raises
limit_if(value, when=NOT_NONE)	when(value)	0 is honored
offset_if(value, when=NOT_NONE)	when(value)
paginate(page, size, max_size=None)	both non-None	1-based, page clamped to ≥ 1, size capped by max_size
apply(fn, when=True)	when	escape hatch: any Select -> Select
build()	always	extract the Select — the terminal step
build_count()	always	the Select for count(*) over the same filters, ignoring paging

Filtering — where_if / exclude_if

Three ways to write a condition:

select_if(User).where_if(name, User.name)              # bare column -> User.name == name
select_if(User).where_if(ages, User.age)               # list value  -> User.age.in_(ages)
select_if(User).where_if(min_age, User.age >= min_age) # any boolean clause, as-is
select_if(User).where_if(min_age, lambda v: User.age >= v)  # callable, gets the value

The bare-column form drops the repeated value in User.name == name:

• a scalar value → column == value (and None → IS NULL);
• a collection (list / tuple / set) → column.in_(value);
• a str/bytes value is always scalar (never iterated into an IN).

Anything that is already a predicate (User.age >= x, .in_(...), ...) is used untouched.

Empty collections: under the default NOT_NONE, where_if([], User.age) applies age IN () (matches nothing). If an empty list should mean "no filter", gate it with when=NOT_EMPTY.

Excluding instead of matching: exclude_if is the exact inverse — same condition forms and when gate, wrapped in NOT (...).

select_if(User).exclude_if(banned_ids, User.id)   # User.id NOT IN (banned_ids)
select_if(User).exclude_if(status, User.status)   # User.status != status
select_if(User).exclude_if(None, User.deleted_at, when=ALWAYS)  # deleted_at IS NOT NULL

Ranges — range_if()

The ?min_x=&max_x= pair, without a lambda per bound.

select_if(User).range_if(min_age, max_age, User.age)   # age >= lo AND age <= hi

Both bounds are inclusive, and each is gated separately, so a one-sided range is just a None:

select_if(User).range_if(None, max_age, User.age)        # age <= max_age
select_if(Post).range_if(after, None, Post.created_at)   # created_at >= after

lo > hi raises ValueError — that query can only return nothing, and it's virtually always a client mistake, so it's a 400 you can catch rather than an unexplained empty page (the same stance as order_if on a bad ?sort=).

The search box — search_if()

select_if(User).search_if(q, User.name, User.email)
# WHERE name LIKE '%q%' OR email LIKE '%q%'   -- columns are alternatives

The term is escaped. % and _ are LIKE metacharacters, so the obvious hand-rolled version is subtly wrong:

User.name.ilike(f"%{q}%")     # q = "%"   -> matches EVERY row
User.name.ilike(f"%{q}%")     # q = "a_c" -> also matches "abc"
select_if(User).search_if(q, User.name)   # both are compared as literal text

You choose where the wildcards go via match=, because %q% can't use a B-tree index but q% can — a real difference on a large table.

match=	SQL	note
CONTAINS (default)	LIKE '%q%'	matches anywhere; can't use a plain index
STARTS_WITH	LIKE 'q%'	prefix search — index-friendly
ENDS_WITH	LIKE '%q'	suffix search
EXACT	= q	no wildcards

from sqla_if import STARTS_WITH

select_if(User).search_if(q, User.name, match=STARTS_WITH)

A matcher is just (column, value) -> clause (the exported Match type), so anything not listed is a lambda — e.g. a Postgres trigram search:

select_if(User).search_if(q, User.name, match=lambda col, v: col.op("%")(v))

Empty search box: under the default NOT_NONE, q="" is not missing — it applies LIKE '%%', which matches every row except those where the column is NULL. If an empty box should mean "no search", pass when=NOT_EMPTY.

Ordering — order_if() / order_by_map()

Two different jobs: order_if takes a sort spec naming columns; order_by_map dispatches a key to arbitrary expressions.

Sort specs

"created_at"          # ascending
"-created_at"         # descending (leading '-')
"+name"               # explicit ascending
"-created_at,name"    # comma separated, applied left to right
["-created_at", "name"]

Column names are validated against a whitelist:

• order_if(sort) — inferred from the columns the SELECT returns.
• order_if(sort, columns=User) — a mapped class or Table.
• order_if(sort, columns=[User, Post]) — several, for a joined query.
• order_if(sort, columns={"name": User.name}) — an explicit mapping (tightest).

Unknown columns raise ValueError — map it to a 400 in your web layer.

With a list, every column also gets a qualified name, and a bare name works only where it's unambiguous:

.order_if(sort, columns=[User, Post])
# ?sort=title      -> ORDER BY posts.title    (only Post has it)
# ?sort=posts.id   -> ORDER BY posts.id
# ?sort=id         -> ValueError: both models have one; the error lists
#                     users.id and posts.id so the caller can pick

Only whitelist columns you can guarantee are joined. ?sort=posts.title on a query that didn't join posts compiles to an ORDER BY over a table absent from the FROM clause, which the database rejects.

If you paginate, also pass tiebreak=.

order_by_map — ordering by arbitrary expressions

order_if sorts by column names. When each sort option maps to a bespoke expression (a window function, a CASE, a .desc() on a computed column), use order_by_map to dispatch by key instead of an if/elif ladder:

# BEFORE
if params.order == "recent":
    query = query.order_by(Post.created_at.desc())
elif params.order == "title":
    query = query.order_by(Post.title.asc(), Post.id.asc())
elif params.order == "rank":
    query = query.order_by(hotness_score.desc())

# AFTER
select_if(Post).order_by_map(params.order, {
    "recent": Post.created_at.desc(),
    "title":  [Post.title.asc(), Post.id.asc()],   # sequence = multi-column
    "rank":   hotness_score.desc(),                 # any expression
})

• Each value is a single order expression or a sequence of them.
• Keys can be strings or enums.
• key is None (or fails when) → skipped (natural default order).
• key present but not in the mapping → ValueError, unless you pass default=<expression>.

.order_by_map(order, mapping, default=Post.id.desc())   # fallback instead of raising

Paging — paginate() / build_count()

builder = (
    select_if(User)
    .where_if(name, User.name)
    .order_if(sort, columns=User, tiebreak=User.id)
    .paginate(page, size, max_size=100)
)

total = session.execute(builder.build_count()).scalar()    # 25
items = session.execute(builder.build()).scalars().all()   # at most `size`

paginate is 1-based and clamps page to ≥ 1. Three things it's easy to get wrong, in order of how much they hurt:

Stable pagination — tiebreak=

Pass tiebreak= whenever you paginate. Ordering by a column with duplicate values leaves the order of the tied rows undefined, so the database may assign them to pages differently on each query — and it does, when the plan changes between two requests. Rows then repeat on one page and never appear on another.

Same query, two plans, every row tied on grade:

SELECT name FROM tied ORDER BY grade LIMIT 2 OFFSET {0,2,4}
  table scan        -> [zoe, yan] [xu, wes] [vic, uma]
  index(grade,name) -> [uma, vic] [wes, xu] [yan, zoe]

A unique column at the end makes the order total, which fixes it:

select_if(User).order_if(sort, columns=User, tiebreak=User.id).paginate(page, size)
# ORDER BY users.age ASC, users.id ASC

tiebreak takes one expression or a list, accepts a direction (tiebreak=User.id.desc()), and won't repeat a column the sort already uses. order_by_map takes it too.

It applies only when the sort does: with no ?sort= there's no ORDER BY at all, and paging is unordered. Use order_by_map(..., default=...) for a baseline ordering.

Capping the page size — max_size=

?size=1000000 otherwise reaches LIMIT untouched:

.paginate(page, size, max_size=100)

It clamps rather than raises, the same as page already does with 0. There's no default cap: adding one would silently truncate results for callers who never asked for it.

The total — build_count()

A list endpoint returns a page and the total it is a page of. build_count() is the companion to paginate: same filters, no paging. Both are extraction steps that hand you a Select: build() for the rows, build_count() for how many there are in total.

limit/offset are dropped — counting them would report the page size (10) instead of the total (25). ORDER BY is dropped too: it cannot change a count, and ordering by a column that isn't selected errors on some backends once wrapped in a subquery.

The statement is wrapped as a subquery rather than having its columns swapped for count(*), so DISTINCT, GROUP BY and HAVING keep counting the rows the query would actually return:

b = select_if(User).join(Post, Post.author_id == User.id)
session.execute(b.build_count()).scalar()              # 3 — joined rows
session.execute(b.distinct().build_count()).scalar()   # 2 — distinct users

Optional joins — join_if()

Several optional filters often need the same join. join_if adds it once, however many of them ask:

(select_if(User)
    .join_if(org_name, Org).where_if(org_name, Org.name)
    .join_if(org_tier, Org).where_if(org_tier, Org.tier))

Done by hand with .apply(lambda s: s.join(Org)), sending both parameters emits JOIN orgs ... JOIN orgs .... Nothing complains while the statement is built — it fails at the database:

sqlite3.OperationalError: ambiguous column name: orgs.name

(PostgreSQL rejects it too — table name "orgs" specified more than once.)

Either filter alone works; only both together break. That's how it reaches production.

target may be a mapped class, a Table, an aliased() construct, or a relationship (User.posts). onclause, isouter and full pass through to Select.join. Dedup reads the statement's FROM clause, so a plain .join() earlier in the chain counts too.

Joining one table twice

An aliased() target is its own FROM element — inspect() resolves it to the alias, not to orgs — which is how you do it on purpose:

billing = aliased(Org)   # a user's org, and the org that gets billed

(select_if(User)
    .join_if(org, Org, Org.id == User.org_id).where_if(org, Org.name)
    .join_if(bill, billing, billing.id == User.bill_id).where_if(bill, billing.name))

# FROM users JOIN orgs          ON orgs.id = users.org_id
#            JOIN orgs AS orgs_1 ON orgs_1.id = users.bill_id
# WHERE orgs.name = ... AND orgs_1.name = ...

SQLAlchemy names the alias orgs_1, which is what keeps the two orgs.name references apart — the absence of exactly that is what makes the accidental double join fail.

Note both joins spell out an onclause. Needing an alias means the two tables have more than one foreign key between them, and SQLAlchemy then can't infer either join (AmbiguousForeignKeysError). It's only inferrable when there's a single unambiguous FK.

A repeat is skipped, not merged: the first call decides the onclause. Joining the same table on two different conditions needs an alias — as SQL requires anyway.

Conflicting join kinds

A repeat asking for a different kind of join raises instead of being skipped:

(select_if(User)
    .join_if(a, Org)                  # JOIN
    .join_if(b, Org, isouter=True))   # ValueError

# orgs is already joined as JOIN, but this asks for LEFT OUTER JOIN. Make the
# join_if calls agree, or join an aliased() copy to get a second, separate join.

Skipping it would keep the inner join and silently drop every row without a match — rows the caller asked to keep. That's a wrong answer with nothing to notice, so it's an error.

The when= predicate — in full

Quick start covers the default. The rest:

1. Built-in predicates

predicate	rule	import
NOT_NONE (default)	value is not None	from sqla_if import NOT_NONE
NOT_EMPTY	not None, and not an empty str/bytes/list/tuple/set/dict	from sqla_if import NOT_EMPTY
TRUTHY	bool(value) — the classic if value:	from sqla_if import TRUTHY
ALWAYS	always True (never skips)	from sqla_if import ALWAYS
IF_SET	value is not UNSET (tri-state, see below)	from sqla_if import IF_SET, UNSET

2. What each predicate does per value — ✅ apply the clause, ⏭️ skip it

value you pass	NOT_NONE (default)	NOT_EMPTY	TRUTHY	ALWAYS
None	⏭️ skip	⏭️ skip	⏭️ skip	✅ apply → IS NULL
0, False	✅ apply	✅ apply	⏭️ skip	✅ apply
"", [], {}	✅ apply	⏭️ skip	⏭️ skip	✅ apply
"alice", 5, True	✅ apply	✅ apply	✅ apply	✅ apply

The default is NOT_NONE (not TRUTHY) on purpose: 0, False and "" are often valid filter values. limit_if(0) must mean LIMIT 0, not "no limit". TRUTHY would silently drop them.

3. Which one do I want?

I want to…	use
Skip only when the value wasn't provided (safe default)	(omit when) → NOT_NONE
Skip empty search boxes, but keep 0 / False	NOT_EMPTY
Skip anything falsy (old-school if value:)	TRUTHY
Always apply — turn a passed None into an IS NULL filter	ALWAYS
Tell "omitted" apart from "explicitly null"	IF_SET + UNSET
Anything else	your own lambda v: ...

.where_if(v, cond)                          # NOT_NONE (default)
.where_if(v, cond, when=NOT_EMPTY)
.where_if(v, cond, when=TRUTHY)
.where_if(v, cond, when=lambda x: x is not None and x > 10)

Filtering for IS NULL

SQLAlchemy compiles column == None to column IS NULL. The *_if machinery already produces that — the only obstacle is the default gate skipping None. Pick the row that matches what you need:

your intent	write	resulting SQL
Always filter IS NULL	select_if(User).where(User.deleted_at.is_(None))	... WHERE deleted_at IS NULL
A passed None should mean IS NULL	select_if(User).where_if(None, User.deleted_at, when=ALWAYS)	... WHERE deleted_at IS NULL
A passed None should mean "no filter"	select_if(User).where_if(None, User.deleted_at)	(no WHERE — default NOT_NONE)

Tri-state: omitted vs. explicitly null vs. a value

In Python "param not provided" and "param is None" are both None, so you can't tell them apart from the value alone. Use the UNSET sentinel as the default and gate with IF_SET:

from sqla_if import UNSET, IF_SET

def search(parent=UNSET):                 # default is UNSET, not None
    return select_if(Node).where_if(parent, Node.parent_id, when=IF_SET)

caller passes	parent is	with when=IF_SET	resulting SQL
(nothing)	UNSET	⏭️ skip	(no filter)
null / None	None	✅ apply	parent_id IS NULL
5	5	✅ apply	parent_id = 5

UNSET is meaningful only with IF_SET. Don't pass it under NOT_NONE (it isn't None, so it would apply and build a nonsensical col = UNSET).

Seeing the SQL

sql_str renders a builder (or a plain Select) as readable SQL; print_sql prints it. Both take a builder directly — no .build() needed.

from sqla_if import print_sql, sql_str

print_sql(select_if(User).where_if(name, User.name))
# SELECT users.id, users.name FROM users WHERE users.name = 'alice'

assert "WHERE users.name" in sql_str(builder)   # handy in tests

argument	default	what it does
dialect	None	Spell the SQL as your database does: "mysql", "postgresql", a Dialect, or an Engine. None uses SQLAlchemy's generic dialect.
literal_binds	True	Inline bound values so the output is copy-pasteable. False renders :name_1 placeholders — use it for types with no literal form (e.g. JSON).
pretty	False	Reindent the SQL across multiple lines.

The rendered SQL is for humans only — it is not escaped for execution, so never feed it back into a database.

Note that SQLAlchemy quotes only identifiers that need it, so dialect= changes the output on reserved words rather than everywhere:

print_sql(select_if(orders), dialect="mysql")   # SELECT `order`.`select` FROM `order`

---

FastAPI

See examples/fastapi_app.py for a full search endpoint over a User/Org join: every helper above in one handler, and the ValueError from order_if / range_if turned into a 400.

uv run --extra examples uvicorn examples.fastapi_app:app --reload
# /users?name=alice&min_age=28&sort=-created_at&page=1&size=10
# /users?org=acme&tier=gold     <- two filters, one join

Development

uv pip install -e ".[dev]"
uv run pytest

License

MIT