SQLAlchemy本身不校验用户输入,安全处理必须靠Pydantic模型+显式数据库约束+参数化查询三者配合;Pydantic需启用from_attributes=True,字段加长度/正则等约束,密码用SecretStr,SQL查询必须参数化,数据库层补unique/index/外键等约束,异步场景注意AsyncSession生命周期和expire_on_commit配置。

直接说结论:SQLAlchemy 本身不校验用户输入,安全处理必须靠 Pydantic 模型 + 显式数据库约束 + 参数化查询三者配合,缺一不可。
Pydantic Schema 必须启用 from_attributes=True
很多人把 Pydantic 模型当“只是个类型提示”,结果在 db.query(User).filter(User.email == user_in.email) 里直接拼接字符串,这是典型误区。正确做法是先用 Pydantic 验证并清洗输入,再传给 SQLAlchemy:
-
from_attributes=True是关键配置,否则UserCreate(**data)无法从 ORM 实例反向构造(比如从数据库读出后返回给前端) - 字段要加
min_length、max_length、regex等约束,例如邮箱字段用EmailStr或正则r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$" - 密码字段必须用
SecretStr包裹,避免日志或调试输出明文
SQLAlchemy 查询必须用参数化,禁用字符串格式化
错误示例:session.execute(f"SELECT * FROM users WHERE email = '{email}'") —— 这等于主动打开 SQL 注入大门。
正确写法只有两种:
FastAPI + Flask 混合部署最佳实践,解决路由定义、API 代理等常见问题,适用于同时运行 FastAPI API 与 Flask 前端的场景。
- ORM 方式:
session.query(User).filter(User.email == email).first()(SQLAlchemy 自动参数化) - 原生 SQL 方式:
session.execute(text("SELECT * FROM users WHERE email = :email"), {"email": email}),:email是占位符,不能写成%s或{email} - 特别注意:
filter_by()只接受关键字参数,不支持表达式,复杂条件必须用filter()
数据库层要补上约束,别全指望应用层
Pydantic 和 SQLAlchemy 都可能被绕过(比如直连数据库、迁移脚本、管理后台),所以表结构本身必须带约束:
-
email字段设为unique=True, index=True,并确保数据库实际生效(SQLite 默认不强制 unique,PostgreSQL/MySQL 才真校验) - 密码哈希后存入,字段长度至少
String(128)(bcrypt 输出约 60 字符,留余量) - 敏感字段如
phone、id_card建议加密存储,用pgcrypto(PostgreSQL)或应用层 AES 加密,别裸存 - 外键必须显式声明
ForeignKey("users.id")并配ondelete="CASCADE"或"RESTRICT",防止脏数据
异步场景下 session 生命周期容易失控
用 sqlalchemy[asyncio] 时,常见错误是把 AsyncSession 当全局变量用,或在依赖注入中忘了 await 关闭:
- 依赖函数必须是
async def get_db(),且用async with AsyncSessionLocal() as session:包裹 - 别在中间件或 background task 里复用 session,每个请求一个独立会话
-
expire_on_commit=False要设,否则commit()后对象属性变None,导致后续取值报DetachedInstanceError - 如果用
session.merge()更新对象,记得await session.refresh(obj)同步最新状态,不然返回的是旧值
最常被忽略的一点:环境变量里的数据库密码和 JWT 密钥绝不能硬编码,也别用 .env 文件提交到 Git —— 即使是私有仓库,也要通过 CI/CD 注入或云服务密钥管理器加载。

















