SQL injection in FastAPI

How it arises in FastAPI

Type hints do real security work here. A handler that declares hero_id: int receives an integer or the client receives a 422 response; no SQL can ride along. That makes it easy to assume the framework validates input in general. It validates shape. q: str, a str field on a Pydantic body model, and a header value are all free text.

The usual routes into the statement:

Async drivers called directly. asyncpg uses PostgreSQL's native $1, $2 placeholders: await conn.fetchrow("SELECT * FROM users WHERE name = $1", name). Code written by someone used to %s often falls back to an f-string because the placeholder syntax is unfamiliar.

A session dependency plus text(). The FastAPI tutorial's database pattern injects a session with Depends and runs session.exec(select(Hero).offset(offset).limit(limit)), which binds values. Reporting endpoints then reach for text() and format into it.

Filter and sort parameters. List endpoints accept sort, order and fields. These are identifiers, so they cannot be bound, and they are declared as str.

Background work. A value validated only as a string is handed to BackgroundTasks.add_task() and used in a query after the response has been sent, away from the handler where a reviewer would look.

Vulnerable and fixed

from fastapi import Depends, FastAPI
from sqlalchemy import text
from sqlalchemy.orm import Session

app = FastAPI()


def get_session():
    raise NotImplementedError


@app.get("/invoices")
def invoices(customer: str, sort: str = "issued_at", session: Session = Depends(get_session)):
    stmt = text(f"SELECT id, amount FROM invoices WHERE customer = '{customer}' ORDER BY {sort}")
    return [dict(r._mapping) for r in session.execute(stmt)]

FastAPI's own idiom fixes the identifier: declare the parameter as an Enum, and a value outside the list is rejected before the function body runs. The value becomes a bind parameter:

from enum import Enum

from fastapi import Depends, FastAPI
from sqlalchemy import text
from sqlalchemy.orm import Session

app = FastAPI()


class InvoiceSort(str, Enum):
    issued_at = "issued_at"
    amount = "amount"


def get_session():
    raise NotImplementedError


@app.get("/invoices")
def invoices(
    customer: str,
    sort: InvoiceSort = InvoiceSort.issued_at,
    session: Session = Depends(get_session),
):
    stmt = text("SELECT id, amount FROM invoices WHERE customer = :customer ORDER BY " + sort.value)
    return [dict(r._mapping) for r in session.execute(stmt, {"customer": customer})]

What FastAPI already does for you

Validation by type hint is the framework's strongest protection and it is on by default: integers, UUIDs, dates and enums arrive parsed or not at all. The documented SQL path through SQLModel and select() binds values. Pydantic models give you one place to tighten a string field with a length limit or a pattern. What the framework cannot know is which of your str parameters will later be placed inside a statement.

What es-python adds

es-python marks what the server reads from the connection and keeps following it after FastAPI and Pydantic have parsed it, through the awaits, into the session dependency or the background task. If request characters are in the statement text when the driver sends it, the sql injection event fires: that label covers asyncpg and SQLAlchemy's text(), which are the two most common choices in FastAPI services. The event carries the sqli impact tag and a trace that ends in your path operation function.

The fixed endpoint behaves differently for its two inputs. customer is bound, so it is never part of the SQL and produces no event. sort is an enum member, yet its characters did arrive in the request; if EyalSec cannot prove them harmless the event remains, graded from its source, and you settle it once with a rule for that call site.

The Block backend network attacks template turns sql injection from async drivers into a RuntimeError raised before the query runs.

Review checklist

1. List every str parameter that reaches a query, including body model fields.
2. Declare sort, order and field selectors as Enum or Literal types.
3. Use $1 placeholders with asyncpg and :name with text().
4. Review functions passed to add_task() as carefully as handlers.
5. Search for text(f" and fetch(f".

References

• Reference: FastAPI documentation: SQL (Relational) Databases (checked 2026-10-12)
• Reference: FastAPI documentation: Path Parameters, predefined values (checked 2026-10-12)
• Reference: FastAPI documentation: Background Tasks (checked 2026-10-12)
• Reference: asyncpg documentation: Usage (checked 2026-10-12)