Preventing SQL Injection in Flask and FastAPI

Microframeworks like Flask and modern asynchronous frameworks like FastAPI dominate Python backend development for microservices, machine learning model APIs, and asynchronous real-time applications. Unlike full-stack frameworks like Django, Flask and FastAPI leave database architectural decisions entirely to developers, most commonly pairing with SQLAlchemy, asyncpg, or databases libraries.

Because developers interact directly with raw SQL or configure database sessions manually, unparameterized query construction frequently introduces critical SQL injection vulnerabilities.

To understand baseline attack mechanics, read our core guide on how SQL injection works and why it remains dangerous. If you also maintain monolithic Python applications, review our guide on preventing SQL injection in Django.

In this tutorial, we cover how to eliminate SQL injection in Flask and FastAPI using modern SQLAlchemy 2.0, safe raw text expressions, and Pydantic data validation.

Where SQL injection occurs in Flask and FastAPI

In Flask, client inputs arrive via request.args or request.json. In FastAPI, inputs arrive via path parameters, query parameters, or Pydantic request bodies. A vulnerability emerges when these inputs are concatenated directly into raw database strings:

# DANGEROUS: Flask route with string formatting
@app.route('/api/users/profile')
def user_profile():
    user_id = request.args.get('id')
    
    # Vulnerable string interpolation:
    query = f"SELECT id, username, email FROM users WHERE id = {user_id}"
    result = db.session.execute(text(query)) # EXPLOITABLE
    return jsonify([dict(row) for row in result])

An attacker passing 1 OR 1=1 can dump every user record. In async FastAPI applications utilizing drivers like asyncpg, unparameterized string formatting poses the exact same risk.

SQLAlchemy 2.0: The golden standard for parameterization

SQLAlchemy 2.0 emphasizes explicit type safety and parameterized statements. The core construct for raw queries is sqlalchemy.text(). When using text(), never format the SQL string; instead, bind parameters using named placeholders (:param):

# SECURE: FastAPI endpoint with SQLAlchemy 2.0 parameterized text()
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
from pydantic import BaseModel, EmailStr
from database import get_async_db

app = FastAPI()

@app.get('/api/users/{user_id}')
async def get_user(user_id: int, db: AsyncSession = Depends(get_async_db)):
    # Parameterized query with named placeholder :id
    stmt = text("SELECT id, username, email FROM users WHERE id = :id AND is_active = :active")
    
    # Values passed strictly as a dictionary mapping
    result = await db.execute(stmt, {"id": user_id, "active": True})
    user = result.mappings().first()
    
    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
        
    return user

In this implementation, the database engine compiles the query template independently. The :id value is transmitted across the binary protocol as literal data, entirely neutralizing injection payloads.

Type-safe ORM queries with SQLAlchemy 2.0 Select

When writing ORM queries rather than raw text, SQLAlchemy's select() construct generates parameterized SQL under the hood:

# SECURE: SQLAlchemy 2.0 Select syntax
from sqlalchemy import select
from models import User

async def find_user_by_email(db: AsyncSession, email: str):
    stmt = select(User).where(User.email == email, User.is_active.is_(True))
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

Using Pydantic for rigorous boundary validation

In FastAPI, Pydantic acts as the first line of defense before parameters ever touch database abstraction layers. Strict type enforcement prevents unexpected payload types from entering execution logic:

from pydantic import BaseModel, Field, constr

class ProductFilterSchema(BaseModel):
    category: str = Field(..., min_length=2, max_length=50)
    min_price: float = Field(default=0.0, ge=0)
    max_price: float = Field(default=10000.0, ge=0)
    
    class Config:
        frozen = True

When FastAPI validates route inputs against ProductFilterSchema, non-numeric prices or malformed strings trigger an immediate HTTP 422 Unprocessable Entity error, preventing suspicious inputs from reaching database sessions.

Handling dynamic clauses safely in Python microframeworks

When users request custom sorting or dynamic filtering, developers often mistakenly concatenate column names:

# DANGEROUS: Column name string interpolation
# stmt = text(f"SELECT * FROM items ORDER BY {user_sort_column} ASC") # VULNERABLE

Database drivers only allow parameter binding for values, not SQL identifiers (column or table names). To implement dynamic sorting securely, use an explicit dictionary lookup:

# SECURE: Identifier whitelist mapping
SORTABLE_COLUMNS = {
    "price": "price",
    "name": "name",
    "created": "created_at"
}

def build_secure_order_query(sort_by: str, direction: str = "desc"):
    column = SORTABLE_COLUMNS.get(sort_by, "created_at")
    dir_clause = "DESC" if direction.lower() == "desc" else "ASC"
    
    # Safe to interpolate because column and dir_clause are validated against strict whitelists
    return text(f"SELECT id, name, price FROM items ORDER BY {column} {dir_clause} LIMIT 50")

For engineering standards that keep Python microservices resilient and maintainable, review our clean code and code review habits guide.

Common mistakes in Flask and FastAPI database layers

Anti-Pattern	Security Consequence	Secure Engineering Fix
f-strings inside text()	Raw SQL statement becomes vulnerable to direct injection	Use :named_param placeholders and pass values via second argument
session.execute(f"...")	Bypasses all driver-level escaping and prepared statement logic	Always wrap raw queries with text() and bind parameters
Interpolating Table / Column Names	Parameter binding does not protect SQL identifiers	Validate identifiers against rigid Python set or dict whitelists
Returning Unhandled DB Exceptions	Leaks database schema, table names, and stack traces to clients	Catch SQLAlchemyError and emit opaque HTTP 500 error responses
Excessive Database User Permissions	Successful injection grants table creation or deletion rights	Configure database connection with least-privilege role credentials

Flask and FastAPI security audit checklist

Verify these security standards before promoting services to production:

• Every raw SQL statement wrapped in text() uses named parameters (:key); zero f-strings or .format().
• FastAPI route parameters and request bodies are validated with strict Pydantic schemas.
• Dynamic table names and sort columns are validated against explicit whitelist dictionaries.
• Database error exceptions (SQLAlchemyError) are intercepted by custom exception handlers returning generic responses.
• Automated SAST tools (such as Bandit or Semgrep) inspect Python code for SQL string concatenations in CI.

Bottom line

Preventing SQL injection in Flask and FastAPI requires adopting modern SQLAlchemy 2.0 conventions. By pairing strict Pydantic request validation with parameterized text() queries or type-safe ORM expressions, you safeguard your microservices from unauthorized database access and data exfiltration.

Sources: SQLAlchemy 2.0 Unified Tutorial on Working with Engines and Connections, FastAPI Security Guidelines, OWASP Python Security Cheat Sheet, CWE-89 (evaluated November 2026).