Executing raw SQL safely with text() and bindparams

Write raw SQL as text("... WHERE email = :email") and pass values as parameters — never formatted into the string — attach types with bindparams() where the driver needs them, and use bindparam(expanding=True) for IN lists; identifiers and syntax cannot be parameters and need an allowlist. This guide belongs to Core vs ORM architecture decisions.

Quick Answer

There are two ways to get a value into raw SQL, and only one of them is a value.

Before — the value formatted into the statement:

from sqlalchemy import text


async def find_by_email(session, email: str):
    sql = f"SELECT id, email FROM customers WHERE email = '{email}'"
    return (await session.execute(text(sql))).mappings().all()

# find_by_email(session, "x' OR '1'='1")
#   → SELECT id, email FROM customers WHERE email = 'x' OR '1'='1'
# Every distinct email is also a distinct statement for the planner to parse.

After — the value as a bound parameter:

from sqlalchemy import String, bindparam, text

FIND_BY_EMAIL = text(
    "SELECT id, email, created_at FROM customers WHERE lower(email) = lower(:email)"
).bindparams(bindparam("email", type_=String))


async def find_by_email(session, email: str):
    result = await session.execute(FIND_BY_EMAIL, {"email": email})
    return result.mappings().all()

For an IN list, one parameter expands into many at execution time:

from sqlalchemy import Integer, bindparam, text

BY_IDS = text("SELECT id, sku FROM products WHERE id IN :ids").bindparams(
    bindparam("ids", type_=Integer, expanding=True)
)

rows = (await session.execute(BY_IDS, {"ids": [1, 2, 3]})).mappings().all()
# ... WHERE id IN ($1, $2, $3)

expanding=True is what makes a list work: without it, the list is passed as a single value and PostgreSQL rejects it.

Execution Context & Async Workflow Integration

text() is not an escape hatch from SQLAlchemy — it is a construct like any other, with a fixed SQL string and a set of named parameters. SQLAlchemy parses the string for :name tokens, records them as bind parameters, and at execution time asks the dialect to rewrite them into the driver's style. For asyncpg that is $1, $2; for psycopg it is %(name)s. That translation is why the same text() runs unchanged on any driver, and why writing driver-native placeholders directly is a mistake — %s reaches asyncpg as a literal percent sign and produces syntax error at or near "%".

Attaching types with bindparams() matters more under asyncpg than under text-based drivers. asyncpg prepares statements server-side, and PostgreSQL must be able to infer a type for every parameter. In a comparison against a typed column it can; in an expression like :value IS NULL or a bare COALESCE(:a, :b) it cannot, and the error is could not determine data type of parameter $1. bindparam("value", type_=String) supplies it.

Types also govern result decoding. text() returns whatever the driver produces, with no type information, which is usually fine — asyncpg already returns datetime, Decimal and dict for the appropriate PostgreSQL types. Where a specific conversion is wanted, text().columns() attaches result types:

from sqlalchemy import DateTime, Integer, text

STATS = text("SELECT id, created_at FROM customers WHERE id = :id").columns(
    id=Integer, created_at=DateTime(timezone=True)
)

Executing raw SQL through the session keeps it inside the session's transaction, which is the usual reason to prefer session.execute(text(...)) over reaching for a connection: a diagnostic query, an advisory lock or a SET LOCAL then participates in the same unit of work as the ORM statements around it.

exec_driver_sql() is a different tool and worth distinguishing. It bypasses SQLAlchemy's parameter handling entirely and passes the string and parameters straight to the driver, so the placeholder style must be the driver's own. It exists for cases where SQLAlchemy's parsing gets in the way — SQL containing colons that are not parameters, for instance, such as PostgreSQL's :: casts in a complex expression. Everywhere else, text() is the safer default.

Shaping results is the last piece. .mappings() gives dictionaries keyed by column name, .scalars() gives the first column, and .all(), .one() and .one_or_none() control cardinality with proper errors — which is better than indexing a list and hoping.

Resolving Warnings, Errors & Common Mistakes

The escaped-colon case catches people writing PostgreSQL casts in raw SQL. SQLAlchemy sees :jsonb in data::jsonb as a parameter named jsonb:

from sqlalchemy import text

# Wrong: ':jsonb' is parsed as a bind parameter.
text("SELECT data::jsonb FROM events")

# Right: escape the colon, or use cast() from Core.
text(r"SELECT data\:\:jsonb FROM events")

Using cast(Event.data, JSONB) from Core avoids the question entirely, which is the general lesson: the more SQL is expressed as constructs, the fewer string-level rules there are to remember.

Identifiers are the other category that cannot be parameterised. A table or column name is part of the statement's structure, so text("SELECT * FROM :table") is not merely unsafe — it is invalid SQL. When a name genuinely has to vary, resolve it through an allowlist and quote it:

from sqlalchemy import text
from sqlalchemy.sql.elements import quoted_name

STAT_TABLES = {"orders": "orders", "customers": "customers", "products": "products"}


async def row_count(session, table: str) -> int:
    name = STAT_TABLES.get(table)
    if name is None:
        raise ValueError(f"unknown table: {table}")
    return await session.scalar(text(f'SELECT count(*) FROM "{quoted_name(name, True)}"'))

The allowlist is what makes it safe; the quoting only handles names with unusual characters. The same reasoning applies to caller-supplied sort keys and filter fields, which is the subject of building dynamic filters and sorting from API query parameters.

Advanced: Mapping Raw SQL Back to ORM Entities

Sometimes the SQL has to be hand-written — a vendor-specific function, a query the planner only gets right one particular way — but the result should still be ORM objects. from_statement() does that:

from sqlalchemy import select, text

from shop.models import Product

SEARCH = text("""
    SELECT products.*
    FROM products
    WHERE products.search_vector @@ websearch_to_tsquery('english', :q)
    ORDER BY ts_rank_cd(products.search_vector, websearch_to_tsquery('english', :q)) DESC
    LIMIT :limit
""")


async def search(session, q: str, limit: int = 20) -> list[Product]:
    stmt = select(Product).from_statement(SEARCH)
    return list(await session.scalars(stmt, {"q": q, "limit": limit}))

The objects come back in the identity map, so they behave like any other loaded entities — including for updates. Two requirements come with it: the SQL must select every column the mapping needs (products.* satisfies that), and loader options for relationships still apply, so eager loading works as usual.

For a query that returns entity columns and extra computed columns, text().columns() names them so the ORM can split the row:

from sqlalchemy import Float, select, text

from shop.models import Product

RANKED = text("""
    SELECT products.*, ts_rank_cd(products.search_vector,
           websearch_to_tsquery('english', :q)) AS rank
    FROM products
    WHERE products.search_vector @@ websearch_to_tsquery('english', :q)
""").columns(*Product.__table__.columns, rank=Float)


async def search_with_rank(session, q: str) -> list[tuple[Product, float]]:
    stmt = select(Product, RANKED.selected_columns.rank).from_statement(RANKED)
    return [(row[0], row[1]) for row in (await session.execute(stmt, {"q": q}))]

Before reaching for either, check whether the query is genuinely inexpressible as constructs. Most PostgreSQL features are available: window functions, CTEs, FILTER, ON CONFLICT, jsonpath, full-text search, LATERAL. The constructs compose — a filter can be added to a select() and not to a text() — carry types, and are checked by type checkers and by SQLAlchemy itself. The genuinely raw cases are narrow: server catalogue queries, a handful of vendor clauses SQLAlchemy does not build such as CYCLE, and EXPLAIN.

Server introspection is the one place raw SQL is unambiguously right, because there is no model to map and no benefit to pretending otherwise:

from sqlalchemy import text

POOL_ACTIVITY = text("""
    SELECT application_name, state, count(*) AS backends,
           max(now() - state_change) AS longest
    FROM pg_stat_activity
    WHERE datname = current_database()
    GROUP BY application_name, state
    ORDER BY backends DESC
""")

Queries like that belong in a diagnostics module, next to the metrics they feed — the observability practices in instrumenting and observing async queries.

Keeping Raw SQL Reviewable

Raw SQL in an application tends to accumulate in the least visible places: a string built in a service function, a query inside a report generator, a diagnostic added during an incident and never removed. Three habits keep it manageable.

Keep statements at module level, as constants. A text() defined once at import is parsed once, is visible to a reader scanning the module, and can be tested on its own. A text() built inside a function from pieces is the shape that eventually grows an f-string:

# shop/sql.py — every raw statement in the service, in one place
from sqlalchemy import Integer, String, bindparam, text

FIND_BY_EMAIL = text(
    "SELECT id, email FROM customers WHERE lower(email) = lower(:email)"
).bindparams(bindparam("email", type_=String))

PRODUCTS_BY_IDS = text("SELECT id, sku FROM products WHERE id IN :ids").bindparams(
    bindparam("ids", type_=Integer, expanding=True)
)

A module like that also makes the inventory auditable: if raw SQL only exists there, reviewing it is reviewing one file.

Ban string formatting mechanically. The failure mode is always an f-string or a % format reaching a text() call, and a lint rule catches it more reliably than review. Ruff's flake8-bandit rules include S608 for hardcoded SQL expressions; a project-specific check can be stricter, rejecting any text() whose argument is not a plain literal:

import ast
import pathlib
import sys

problems = []
for path in pathlib.Path("shop").rglob("*.py"):
    tree = ast.parse(path.read_text(), filename=str(path))
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.Call)
            and getattr(node.func, "id", None) == "text"
            and node.args
            and not isinstance(node.args[0], ast.Constant)
        ):
            problems.append(f"{path}:{node.lineno} text() with a non-literal argument")

if problems:
    print("\n".join(problems))
    sys.exit(1)

Test the statement, not just the function. A raw statement's correctness is in its SQL, so a test that executes it against a real PostgreSQL with known fixture rows is worth more than a mock. It also catches the parameter-name mismatches and type-inference errors that only appear at execution:

import pytest

from shop.sql import PRODUCTS_BY_IDS


@pytest.mark.asyncio
async def test_products_by_ids_expands(session, product_factory):
    a, b = await product_factory(sku="A"), await product_factory(sku="B")
    rows = (await session.execute(PRODUCTS_BY_IDS, {"ids": [a, b]})).mappings().all()
    assert {row["sku"] for row in rows} == {"A", "B"}


@pytest.mark.asyncio
async def test_products_by_ids_handles_an_empty_list(session):
    rows = (await session.execute(PRODUCTS_BY_IDS, {"ids": []})).mappings().all()
    assert rows == []

The empty-list test earns its place: an expanding parameter with no values renders a construct PostgreSQL accepts, and code that assumed at least one id would otherwise fail only in production.

Frequently Asked Questions

Related

• Core vs ORM Architecture Decisions — The parent guide: choosing a layer, and mixing them.
• Mapping classes to existing tables with reflection — When the schema is not yours to model.
• Building dynamic filters and sorting from API query parameters — Allowlists for identifiers and operators.
• Reading EXPLAIN output for a SQLAlchemy query — The most common legitimate use of raw SQL.