FastAPI + SQLAlchemy 默认防SQL注入,关键在禁用字符串拼接、坚持参数化查询、白名单校验表名/字段名、Pydantic显式约束输入。

只要不用字符串拼接构造 SQL,FastAPI + SQLAlchemy 默认就是防注入的;真正危险的操作,往往藏在你自以为“灵活”的地方。
用 ORM 查询方法时,SQLAlchemy 自动参数化
SQLAlchemy 的 filter()、filter_by()、where() 等方法底层全部走参数化绑定,用户输入不会被解析为 SQL 代码。
- ✅ 安全写法:
session.query(User).filter(User.email == email_input)或select(User).where(User.email == email_input) - ❌ 危险写法:
session.query(User).filter(f"email = '{email_input}'")或text("SELECT * FROM users WHERE email = '" + email_input + "'") - 异步场景下
AsyncSession行为一致:await session.execute(select(User).where(User.email == email_input))同样安全
手写原生 SQL 必须用 text() + 命名参数
当你必须用 text() 执行原始 SQL(比如复杂窗口函数、动态排序),参数只能通过命名占位符传入,不能拼接。
FastAPI + Flask 混合部署最佳实践,解决路由定义、API 代理等常见问题,适用于同时运行 FastAPI API 与 Flask 前端的场景。
- ✅ 正确:
text("SELECT * FROM users WHERE status = :status AND created_at > :since"),再传{"status": "active", "since": dt} - ❌ 错误:
text(f"SELECT * FROM users WHERE status = '{status}'")—— 这直接绕过所有防护 -
IN子句要特别注意:值必须是tuple,且用:in_list占位:text("WHERE id IN :in_list"),传参时{"in_list": tuple(ids)}
表名、字段名等结构部分无法参数化,必须白名单校验
SQLAlchemy 不允许对表名、列名、ORDER BY 字段做参数化——因为它们属于 SQL 结构,不是数据。这部分一旦由用户控制,就必须硬性限制。
- 例如多租户场景下根据
tenant_id切换表前缀,得先校验:if schema_name not in ["public", "tenant_a", "tenant_b"]: raise HTTPException(400) - 搜索字段排序(如
?sort=created_at&order=desc)也要白名单:if sort_field not in ["created_at", "name", "email"]: sort_field = "created_at" - 别用
getattr(User, sort_field)直接反射字段,除非你确认sort_field已过滤
Pydantic 模型不是银弹,字段约束得写实
Pydantic 能帮你挡掉明显非法输入,但它不自动清理或截断;如果字段只声明 str,攻击者仍可传入超长 payload 或嵌套结构触发下游逻辑异常。
- 显式加约束:
username: str = Field(min_length=2, max_length=50, pattern=r'^[a-zA-Z0-9_]+$') - 语义类型优先:
email: EmailStr、url: HttpUrl(需安装pydantic[email]) - 搜索字段(如
q: str)建议配合Query(strip_whitespace=True, max_length=100)
最容易被忽略的点是:**动态 SQL 结构 + 未校验的用户输入 = 注入温床**。不是 ORM 不够强,而是人总想“临时拼一下”,结果把门打开了。

















