SQL Injection in 2026: What SQLAlchemy Still Can't Protect Against

The Assumption That Breaks

The safe version of a SQLAlchemy filter looks like this:

user = User.query.filter(User.email == email).first()

email is never concatenated into the query string. It's bound as a parameter. The database engine handles escaping. This is correct, and it covers the majority of what a Flask app does.

The problem is that SQLAlchemy also provides direct SQL execution for cases where the ORM expression language isn't enough. The line between "parameterized ORM query" and "raw SQL string with substitution" is easy to cross without noticing — especially in a codebase where both patterns coexist.

Pattern 1: text() With String Formatting

SQLAlchemy's text() function is the escape hatch for raw SQL:

from sqlalchemy import text

result = db.session.execute(
    text(f"SELECT * FROM users WHERE email = '{email}'")
)

This is injection. email is a request value, it goes directly into the query string, and SQLAlchemy will execute whatever you put there. The parameterized version:

result = db.session.execute(
    text("SELECT * FROM users WHERE email = :email"),
    {"email": email}
)

SQLAlchemy can't distinguish text("safe query") from text(f"query with {user_input}"). It executes the string. The :param named binding is what makes parameterization work — not the fact that you're using text().

This pattern shows up most in complex queries that the ORM makes awkward: window functions, CTEs, complex lateral joins. The query starts safe. Six months later, someone adds f"AND region = '{region}'" to the end of a working string, and the injection surface is in a query that otherwise looks fine.

Grep your codebase for text(f" and text(" followed by .format( or %. Those are the sites worth auditing.

Pattern 2: order_by() With User-Controlled Column Names

This one looks different from injection but isn't:

sort_column = request.args.get('sort', 'created_at')
sort_dir = request.args.get('dir', 'asc')

users = User.query.order_by(text(f"{sort_column} {sort_dir}")).all()

The filter conditions on User.query are parameterized. order_by(text(...)) is not, and it can't be — column names aren't data. SQL doesn't support parameterizing what columns a query sorts by. The database needs the column name at query planning time, not execution time.

So if you're building a dynamic sort — which is almost every table in any admin UI — the column name has to enter the SQL string. Without a guard, the injection path is:

/users?sort=created_at;DROP+TABLE+users;--&dir=asc

The fix is an allowlist:

ALLOWED_SORT_COLUMNS = {'created_at', 'username', 'email', 'last_login'}
ALLOWED_SORT_DIRS = {'asc', 'desc'}

if sort_column not in ALLOWED_SORT_COLUMNS:
    sort_column = 'created_at'
if sort_dir not in ALLOWED_SORT_DIRS:
    sort_dir = 'asc'

users = User.query.order_by(text(f"{sort_column} {sort_dir}")).all()

A cleaner option is to map frontend names to ORM column objects and never construct a string from user input:

from sqlalchemy import asc, desc

SORT_MAP = {
    'created': User.created_at,
    'name': User.username,
    'email': User.email,
}

col = SORT_MAP.get(sort_column, User.created_at)
direction = desc if sort_dir == 'desc' else asc
users = User.query.order_by(direction(col)).all()

The column reference is a Python object. No string concatenation. The API name (created) can differ from the schema column name (created_at) without leaking schema details to the client.

Pattern 3: Dynamic Column Selection

Less common but worth auditing. If you're exposing a fields parameter that controls which columns to return:

columns = request.args.getlist('fields')
entities = [getattr(User, col) for col in columns]
result = User.query.with_entities(*entities).all()

getattr(User, col) raises AttributeError for columns that don't exist on the model. That's not injection, but if you're catching AttributeError anywhere in the call chain and silently continuing, you'll miss the error and expose confusing behavior.

The worse version:

entities = [text(col) for col in columns]
result = User.query.with_entities(*entities).all()

That's injection through column selection. Same fix — allowlist and map to ORM objects:

FIELD_MAP = {
    'id': User.id,
    'username': User.username,
    'email': User.email,
    'created_at': User.created_at,
}

requested = request.args.getlist('fields')
entities = [FIELD_MAP[f] for f in requested if f in FIELD_MAP]
if not entities:
    entities = [User.id, User.username]

result = User.query.with_entities(*entities).all()

No string ever touches the query. The client gets exactly the columns they asked for that exist in the allowlist.

Pattern 4: Alembic Migration Scripts

Migrations run offline, but they run with full database access during deploys. Alembic's op.execute() accepts a raw string:

def upgrade():
    op.execute("UPDATE users SET plan = 'free' WHERE plan IS NULL")

That string is in version control and has no injection risk. But migrations that build queries dynamically are different:

def upgrade():
    conn = op.get_bind()
    for table_name in config.get_main_option("source_tables").split(","):
        conn.execute(text(f"INSERT INTO archive SELECT * FROM {table_name}"))

If source_tables in alembic.ini or an environment variable can be influenced from outside the deployment pipeline, every table name that passes through text(f"...{table_name}...") is an injection site — one that runs with whatever privileges the migration user has.

Migrations that need real table names should validate against the live schema, not a config value:

from sqlalchemy import inspect

def upgrade():
    conn = op.get_bind()
    inspector = inspect(conn)
    existing_tables = set(inspector.get_table_names())

    for table_name in get_expected_source_tables():
        if table_name not in existing_tables:
            raise ValueError(f"Table {table_name!r} not found in schema")
        conn.execute(text(f"INSERT INTO archive SELECT * FROM {table_name}"))

table_name still goes into the SQL string, but it's now validated against what the database actually contains. A config value that doesn't match a real table causes the migration to fail loudly instead of executing arbitrary SQL.

The Coverage Map

The ORM covers data values. You cover structure — column names, table names, SQL keywords. Anything built from user input that isn't a data value needs an allowlist before it touches a query string.

The failure mode that catches teams isn't someone writing obvious injection. It's a developer who knows the ORM is safe, adds a text() call for one edge case, and doesn't notice that text() with an f-string is a different code path. The query still looks like ORM code from a distance. The vulnerability is in the part that reads like a minor implementation detail.