Building dynamic filters and sorting from API query parameters

Resolve every caller-supplied field and operator through an explicit allowlist that maps a public name to a column and the operators it permits, build each condition as a bound parameter, and end every sort with a unique tiebreaker — never getattr(Model, request_field) and never a text() fragment built from input. This guide belongs to hybrid properties, column properties and SQL expressions.

Quick Answer

The convenient version resolves attribute names dynamically, which exposes every attribute on the class and puts caller input into the SQL.

Before — attribute lookup and an operator from the request:

from sqlalchemy import select, text

from shop.models import Product


def search(params: dict):
    stmt = select(Product)
    for field, value in params.items():
        column = getattr(Product, field)              # any attribute, including metadata
        stmt = stmt.where(column == value)
    if "sort" in params:
        stmt = stmt.order_by(text(params["sort"]))    # caller-controlled SQL
    return stmt

After — an allowlist of fields, operators and sort keys:

from dataclasses import dataclass
from typing import Any, Callable

from sqlalchemy import select
from sqlalchemy.sql.elements import ColumnElement

from shop.models import Product


@dataclass(frozen=True)
class Filterable:
    column: Any
    operators: frozenset[str]


FILTERS: dict[str, Filterable] = {
    "sku": Filterable(Product.sku, frozenset({"eq", "in"})),
    "brand": Filterable(Product.brand, frozenset({"eq", "in", "contains"})),
    "price": Filterable(Product.price_cents, frozenset({"eq", "gt", "gte", "lt", "lte"})),
    "in_stock": Filterable(Product.in_stock, frozenset({"eq"})),
}

SORTS = {"sku": Product.sku, "price": Product.price_cents, "created": Product.created_at}

OPERATORS: dict[str, Callable[[Any, Any], ColumnElement[bool]]] = {
    "eq": lambda c, v: c == v,
    "ne": lambda c, v: c != v,
    "gt": lambda c, v: c > v,
    "gte": lambda c, v: c >= v,
    "lt": lambda c, v: c < v,
    "lte": lambda c, v: c <= v,
    "in": lambda c, v: c.in_(v),
    "contains": lambda c, v: c.ilike(f"%{escape_like(v)}%", escape="\\"),
}


def escape_like(value: str) -> str:
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")

Every name a caller can send is in FILTERS or SORTS, every operator is in OPERATORS, and every value arrives as a bound parameter. Anything else is rejected before a statement is built.

Execution Context & Async Workflow Integration

SQLAlchemy already protects the values in a query: column == value produces a bound parameter, whatever the value contains. What it cannot protect is the structure — which column, which operator, which sort key — because that is chosen by your code. A dynamic endpoint moves those choices to the caller, so the code's job is to map caller strings onto a closed set of known structures.

getattr(Model, name) fails at exactly that job. A mapped class's namespace contains far more than its filterable columns: relationships, hybrids, metadata, registry, __table__, private helpers, and anything a mixin added. A caller sending field=metadata reaches an object that is not a column at all, and the resulting TypeError is a 500 that leaks a traceback. A caller sending the name of a column the API never meant to expose — an internal risk score, a password hash — gets a working filter over it, which is an information-disclosure bug even when the column is never returned: comparing it repeatedly reveals its value.

text() is the other trap, and a worse one. order_by(text(params["sort"])) places caller input directly into the statement, so price; DROP TABLE products is a syntax error and (SELECT ...) is a subquery of the caller's choosing. There is never a reason to build text() from request data; text() is for SQL you wrote.

The allowlist approach turns both into validation errors. It also gives the endpoint a documented contract for free: the keys of FILTERS are the filters the API offers, and the operator sets say what each one supports — which is exactly what the OpenAPI schema should describe.

Building the statement is then a loop over validated input:

from sqlalchemy import Select, select

from shop.models import Product


def build_query(filters: list[tuple[str, str, object]]) -> Select:
    stmt = select(Product)
    for field, operator, value in filters:
        spec = FILTERS.get(field)
        if spec is None:
            raise ValueError(f"unknown filter field: {field}")
        if operator not in spec.operators:
            raise ValueError(f"operator {operator!r} is not allowed on {field!r}")
        stmt = stmt.where(OPERATORS[operator](spec.column, value))
    return stmt

Under async this is all statement construction, so nothing here is awaited; the built statement goes to await session.execute(stmt) like any other. The one async-specific concern is the result: an endpoint that returns ORM objects whose relationships were not loaded will raise on serialisation, so the loader options a dynamic query needs must be decided by the endpoint, not by the caller. Returning selected columns rather than entities avoids the question entirely, and is usually the better shape for a filterable list endpoint.

Resolving Warnings, Errors & Common Mistakes

Type coercion is the part an allowlist alone does not solve, and it is what makes a validation library worth using. Declaring the request shape once gives typed values, rejects unknown keys, and produces a documented schema:

import datetime as dt
from typing import Annotated, Literal

from pydantic import BaseModel, Field


class ProductQuery(BaseModel, extra="forbid"):
    sku: str | None = None
    brand: str | None = None
    brand_contains: Annotated[str, Field(max_length=64)] | None = None
    price_gte: Annotated[int, Field(ge=0)] | None = None
    price_lte: Annotated[int, Field(ge=0)] | None = None
    created_after: dt.date | None = None
    sort: Literal["sku", "price", "-price", "created", "-created"] = "-created"
    limit: Annotated[int, Field(ge=1, le=100)] = 20
    cursor: str | None = None

extra="forbid" rejects parameters the endpoint does not support instead of ignoring them, which turns a client's typo into an immediate error rather than a silently unfiltered result set. Fixed field names like price_gte also remove the need for a caller-supplied operator string: the operator is part of the name, and the name is in a closed set.

The escape detail is worth spelling out because it is easy to get half right. ilike("%" + value + "%") with an unescaped value lets a caller send % and match every row — cheap for them, a full table scan for you — and _ to match any single character. Escaping the escape character first, then % and _, and telling the database which character is the escape, is the complete fix.

Advanced: Relationship Filters, Sorting and Deep Pages

Filters that reference related tables are where a naive builder goes wrong quietly. Adding a join per filter duplicates parent rows for collection relationships, which breaks counts and paging, and it silently changes the meaning of two filters: joined conditions must be satisfied by the same related row.

from sqlalchemy import select

from shop.models import Product, Review, Tag

# Products tagged "usb" AND tagged "hdmi" — two different tag rows.
stmt = (
    select(Product)
    .where(Product.tags.any(Tag.name == "usb"))
    .where(Product.tags.any(Tag.name == "hdmi"))
)

# Products with at least one review of 5 stars.
stmt = select(Product).where(Product.reviews.any(Review.rating == 5))

Each any() is an independent EXISTS, so the row count is unchanged and each condition can be met by a different related row. A join is right only when columns from the related table are also selected or sorted on — and then it should be an explicit aliased() join so two filters over the same relationship cannot collide.

Sorting needs a total order. Any sort key with duplicate values leaves rows in an undefined relative order, and paging over an undefined order shows some rows twice and skips others:

from sqlalchemy import nulls_last

DIRECTIONS = {"sku": Product.sku, "price": Product.price_cents, "created": Product.created_at}


def apply_sort(stmt, sort: str):
    descending = sort.startswith("-")
    column = DIRECTIONS[sort.lstrip("-")]
    ordered = column.desc() if descending else column.asc()
    # NULLs placed explicitly, and a unique tiebreaker so the order is total.
    return stmt.order_by(nulls_last(ordered), Product.id.desc() if descending else Product.id)

The tiebreaker then does double duty, because it is what makes keyset pagination possible. Instead of OFFSET, which makes the database count and discard every skipped row, the next page is a comparison against the last row of the previous one:

from sqlalchemy import tuple_

def apply_cursor(stmt, last_price: int | None, last_id: int | None):
    if last_price is None:
        return stmt
    return stmt.where(tuple_(Product.price_cents, Product.id) < (last_price, last_id))

A composite index on (price_cents, id) serves that comparison directly, so page one thousand costs the same as page one. The cursor should be opaque to clients — a signed or encoded blob rather than raw values — so the contract stays yours to change; the full treatment is in paginating large result sets with keyset pagination.

Two final safeguards belong on any endpoint like this. Cap the IN list length, because a ten-thousand-element IN is both slow to plan and a cheap way to make your database work hard. And set a statement_timeout for the request path, so a filter combination nobody anticipated fails in two seconds instead of occupying a connection for minutes.

Exposing Hybrids and Derived Values Safely

An allowlist maps names to SQL expressions, and a column is only the simplest kind. That means the same mechanism can expose a hybrid property, a column_property, or an expression built for the occasion — with the caller never knowing the difference.

from sqlalchemy import func

from shop.models import Product

FILTERS = {
    # A plain column.
    "price": Filterable(Product.price_cents, frozenset({"eq", "gt", "gte", "lt", "lte"})),
    # A hybrid property: its expression half renders the SQL.
    "on_sale": Filterable(Product.on_sale, frozenset({"eq"})),
    # A derived count mapped with column_property.
    "review_count": Filterable(Product.review_count, frozenset({"gt", "gte", "lt", "lte"})),
    # An expression built here, not on the model.
    "name_ci": Filterable(func.lower(Product.name), frozenset({"eq", "contains"})),
}

Three cautions come with that power.

Indexability. A filter on a plain indexed column is an index lookup; a filter on func.lower(name) needs an expression index on exactly that expression, and a filter on a correlated column_property cannot use a child-table index to pre-select parents at all. An allowlist entry is therefore a performance commitment: every operator it permits is a query someone will run. Check the plan for each one, as in reading EXPLAIN output for a SQLAlchemy query, before adding it.

Hybrids need their expression half. A hybrid with only a Python getter cannot be filtered on: SQLAlchemy will evaluate the getter at class level and produce either an error or a meaningless clause. Writing hybrid properties that work in Python and SQL covers the pairing; the allowlist should only ever reference hybrids that have both halves.

Global criteria still apply. If the model uses soft deletion with a do_orm_execute listener, dynamic filters compose with it automatically — the listener adds its predicate to whatever statement the builder produced. That is the desired behaviour, and it is worth a test, because a filtered endpoint returning soft-deleted rows is exactly the leak the listener exists to prevent.

Finally, keep the allowlist and the API documentation generated from one source. A dictionary of names to expressions and operator sets can produce the OpenAPI parameter list directly, which means the documented filters and the implemented filters cannot drift apart — and a filter that is removed from the model stops being offered in the same commit.

Frequently Asked Questions

Related

• Hybrid Properties, Column Properties and SQL Expressions — The parent guide: derived values a filter can reference.
• Writing hybrid properties that work in Python and SQL — Hybrids need an expression half before a filter can use them.
• Filtering soft-deleted rows with with_loader_criteria — Global criteria that compose with caller-supplied filters.
• Paginating large result sets with keyset pagination — Deep pages at constant cost, using the sort tiebreaker.