1. 从一次线上慢查询说起：Django 原生 SQL 参数化到底怎么写才安全
Django 的 ORM 用起来确实省心，但真到了复杂报表、多表聚合、窗口函数、EXPLAIN调优这些场景，ORM 生成的 SQL 往往又长又难控。这时候很多人会退回原生 SQL，用connection.cursor()或Model.objects.raw()直接执行。问题也随之而来：参数怎么传才不会被注入？%s和%(name)s有什么区别？raw()返回的模型对象为什么字段对不上？线上和本地行为不一致时又该从哪里查？
这篇就围绕「Django 中使用原生 SQL（带参数）的方法」这个主题，把raw()、cursor()两种写法的参数化细节、注入防护、EXPLAIN验证动作讲清楚。同时结合 TaoToken 统一 Key 的 API 通道，演示在同一个项目里调用多个大模型做 SQL 审查、慢查询解释时，鉴权配置怎么统一管理。适合已经写过 Django 视图、想安全落地原生 SQL 的开发者，也适合正在做多模型调试、需要一套统一 Key 的同学。
核心检索词先摆出来：Django 原生 SQL 参数化查询，指的是通过cursor.execute(sql, params)或raw(sql, params)把用户输入作为独立参数交给数据库驱动，而不是用字符串拼接塞进 SQL 文本。它能防注入、能复用执行计划，适合谁？适合所有需要在 Django 里手写 SQL 又不想背安全锅的人。
我试过在报表接口里直接拼 f-string，本地跑得好好的，上线第二天就被安全扫描标红。后来改成参数化，配合 TaoToken 的模型对话做 SQL 复核，才算把这条链路理顺。下面按步骤拆。
2. TaoToken 前置准备：统一 Key 与多模型鉴权配置
在讲 SQL 之前，先把 TaoToken 这条通道配好，因为后面第五节的 SQL 审查、慢查询解释都要用它。TaoToken 是一个统一 API 通道，官网在 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 入口是 https://taotoken.net/api 。它的价值在于：你只需要一个 Key，就能在同一个项目里切换不同模型，不用为每个模型单独维护一套鉴权和 Base URL。
第一步，去控制台创建 API Key。打开 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，登录后进入 API Keys 页面，新建一个 Key 并复制保存。这个 Key 就是后面所有模型调用的统一凭证，建议按环境拆开，本地一个、线上一个，方便轮换。
第二步，确认你要用的模型 ID。在模型对话页面 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 可以看到当前可用的模型列表，把你要做 SQL 审查的模型 ID 记下来。不同模型对 SQL 方言的理解有差异，建议选一个对 PostgreSQL/MySQL 语法熟悉的。
第三步，把配置写进 Django 的 settings。这里给一个可复制的片段，路径按你自己的项目结构调整。注意 Key 不要硬编码进仓库，用环境变量读取：
# settings.py import os TAOTOKEN_BASE_URL = "https://taotoken.net/api" TAOTOKEN_API_KEY = os.environ.get("TAOTOKEN_API_KEY", "") TAOTOKEN_DEFAULT_MODEL = os.environ.get("TAOTOKEN_MODEL", "claude-sonnet-4-20250514") # 如果你用 OpenAI 兼容的 SDK，可以这样组织 TAOTOKEN_CLIENT_CONFIG = { "base_url": TAOTOKEN_BASE_URL, "api_key": TAOTOKEN_API_KEY, "default_model": TAOTOKEN_DEFAULT_MODEL, "timeout": 60, }如果你更习惯用 TOML 管理配置，可以放在项目根目录的config.toml：
[taotoken] base_url = "https://taotoken.net/api" api_key_env = "TAOTOKEN_API_KEY" default_model = "claude-sonnet-4-20250514" timeout = 60然后在代码里读取。这样做的目的是把「鉴权」和「业务 SQL」解耦：SQL 参数化是数据库层的事，模型调用是应用层的事，两者通过统一 Key 串起来，但互不污染。
第四步，验证 Key 是否可用。写一个最小的请求脚本，确认能拿到模型返回：
# scripts/check_taotoken.py import os from openai import OpenAI client = OpenAI( base_url="https://taotoken.net/api", api_key=os.environ["TAOTOKEN_API_KEY"], ) resp = client.chat.completions.create( model=os.environ.get("TAOTOKEN_MODEL", "claude-sonnet-4-20250514"), messages=[{"role": "user", "content": "回复 ok 两个字母即可"}], ) print(resp.choices[0].message.content)跑通这一步，说明 Base URL、Key、Model ID 三件套都对上了。后面第五节我们会把慢 SQL 丢给这个通道做解释。如果你打算长期做编码和 Agent 类任务，可以了解下 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ，它更适合高频调用场景。
3. 可复制配置：cursor() 与 raw() 的参数化写法对照
这一节是全文的技术核心，给出可以直接抄进项目的配置和代码。先明确一个原则：永远不要把用户输入拼进 SQL 字符串。Django 的数据库驱动支持两种占位符风格，%s（位置参数）和%(name)s（命名参数），两者都能防注入，区别在于可读性和复用性。
先看cursor()的位置参数写法。这是最贴近原生 DB-API 的方式：
# views.py from django.db import connection from django.http import JsonResponse def get_action_by_id(request): action_id = request.GET.get("id") if not action_id: return JsonResponse({"error": "missing id"}, status=400) with connection.cursor() as cursor: cursor.execute( "SELECT id, name, status FROM pcassistant_action WHERE id = %s", [action_id], ) rows = cursor.fetchall() return JsonResponse({"rows": rows}, safe=False)注意execute的第二个参数是列表或元组，驱动会负责转义和类型处理。with connection.cursor()会自动关闭游标，比手动cursor.close()更稳。
再看命名参数写法，适合参数多、顺序容易搞混的场景：
def search_actions(request): keyword = request.GET.get("kw", "") status = request.GET.get("status", "active") with connection.cursor() as cursor: cursor.execute( """ SELECT id, name, status FROM pcassistant_action WHERE name LIKE %(kw)s AND status = %(status)s """, {"kw": f"%{keyword}%", "status": status}, ) rows = cursor.fetchall() return JsonResponse({"rows": rows}, safe=False)命名参数的好处是 SQL 里的%(kw)s和字典 key 一一对应，改顺序不影响结果。LIKE的通配符%要放在参数值里，不能写在 SQL 文本里，否则又变成拼接了。
接下来是raw()的写法。raw()返回的是模型实例，适合你希望结果直接映射到 Model 的场景：
from django.db import models class PcAssistantAction(models.Model): name = models.CharField(max_length=128) status = models.CharField(max_length=32) class Meta: db_table = "pcassistant_action" def get_actions_raw(request): action_id = request.GET.get("id") qs = PcAssistantAction.objects.raw( "SELECT id, name, status FROM pcassistant_action WHERE id = %s", [action_id], ) data = [{"id": obj.id, "name": obj.name, "status": obj.status} for obj in qs] return JsonResponse({"rows": data}, safe=False)raw()的第二个参数同样是参数列表，支持列表和字典两种。它有个坑：SQL 里必须包含主键字段，否则 Django 无法构造模型实例，会报InvalidQuery。另外raw()返回的 QuerySet 支持__iter__，但不支持filter()链式调用，别拿它当普通 QuerySet 用。
如果你需要把结果转成字典列表，cursor()方式可以这样处理：
def rows_to_dicts(cursor): columns = [col[0] for col in cursor.description] return [dict(zip(columns, row)) for row in cursor.fetchall()]这个辅助函数在报表接口里很实用，cursor.description会给出列名，配合zip就能把元组转成字典。
关于IN查询的转义问题，原始资料里提到「SQL 中带引号需要转义」，其实参数化之后不需要手动转义引号，但IN的子项数量是动态的，需要动态生成占位符：
def get_actions_in(request): ids = request.GET.getlist("id") # ?id=1&id=2&id=3 if not ids: return JsonResponse({"rows": []}) placeholders = ",".join(["%s"] * len(ids)) sql = f"SELECT id, name FROM pcassistant_action WHERE id IN ({placeholders})" with connection.cursor() as cursor: cursor.execute(sql, ids) rows = cursor.fetchall() return JsonResponse({"rows": rows}, safe=False)这里placeholders是根据ids长度生成的，%s的数量和参数数量严格对应，用户输入本身仍然走参数通道，所以是安全的。唯一需要小心的是ids的长度，太长会触发数据库的占位符上限，建议做一次长度校验。
配置层面，如果你用 Django 的DATABASES多库，记得cursor()默认走default，要指定库用connections["replica"]。另外开启CONN_MAX_AGE复用连接时，长事务里的游标要及时关闭，避免连接泄漏。
4. 验证请求与成功结果：EXPLAIN 与接口实测
写完代码不能只看「没报错」，要验证两件事：参数确实生效了，SQL 确实走了索引。这一节给出可执行的验证动作。
先验证参数化是否生效。启动开发服务器，用 curl 打接口：
curl "http://127.0.0.1:8000/get_action_by_id?id=10"预期返回类似：
{"rows": [[10, "sync_task", "active"]]}如果传一个带引号或特殊字符的参数，比如?id=10' OR '1'='1，参数化写法会把它当成普通字符串去匹配，返回空结果，而不是把整张表查出来。这就是防注入的直接证据。你可以对比一下拼接写法的行为，但别在线上试。
再验证执行计划。用EXPLAIN看 SQL 是否命中索引：
def explain_action_query(request): action_id = request.GET.get("id") with connection.cursor() as cursor: cursor.execute( "EXPLAIN SELECT id, name FROM pcassistant_action WHERE id = %s", [action_id], ) plan = cursor.fetchall() return JsonResponse({"plan": plan}, safe=False)MySQL 下返回的是逐行计划，PostgreSQL 下可以加ANALYZE看实际耗时：
EXPLAIN ANALYZE SELECT id, name FROM pcassistant_action WHERE id = %s;重点看type是不是const/ref，key是不是主键或索引名，rows扫描行数是不是接近 1。如果看到ALL全表扫描，说明索引没走对，要么字段类型不匹配，要么函数包裹了索引列。
把慢查询丢给 TaoToken 做解释也是个好办法。把EXPLAIN的输出贴进模型对话，让它帮你判断瓶颈：
def ask_model_about_plan(plan_text): from openai import OpenAI import os client = OpenAI( base_url="https://taotoken.net/api", api_key=os.environ["TAOTOKEN_API_KEY"], ) resp = client.chat.completions.create( model=os.environ.get("TAOTOKEN_MODEL", "claude-sonnet-4-20250514"), messages=[ {"role": "system", "content": "你是数据库优化助手，只输出可执行的建议。"}, {"role": "user", "content": f"解释这个执行计划并给出索引建议：\n{plan_text}"}, ], ) return resp.choices[0].message.content实测下来，把EXPLAIN结果和表结构一起给模型，它给出的索引建议比单看 SQL 准得多。注意别把生产库的真实数据贴进去，用脱敏后的表结构即可。
成功结果的判断标准有三条：接口返回结构符合预期、特殊字符参数不改变查询语义、EXPLAIN显示走了索引。三条都过，这个原生 SQL 才算安全落地。
5. 本篇常见错排查：401、local proxy failed、reading choices 与 OAuth
这一节对照真实报错，把踩过的坑列出来。先说数据库侧的，再说 TaoToken 通道侧的。
数据库侧最常见的错误是django.db.utils.ProgrammingError: not enough arguments for format string。原因通常是 SQL 里写了%s但execute没传参数，或者参数数量对不上。检查execute(sql, params)的params是不是列表/元组，长度和%s数量是否一致。另一个变体是TypeError: format string argument must be a mapping，说明你用了%(name)s但传的是列表，改成字典即可。
raw()的典型报错是django.db.models.query_utils.InvalidQuery: Raw query must include the primary key。解决方法是 SQL 的SELECT里显式带上主键字段，比如SELECT id, name FROM ...，别用SELECT *在某些驱动下也可能漏掉主键映射。
TaoToken 通道侧，第一个高频报错是401 Unauthorized。原因一般是 Key 没读到、Key 失效、或者 Base URL 写错。排查顺序：确认环境变量TAOTOKEN_API_KEY在当前 shell 里能echo出来；确认base_url是https://taotoken.net/api而不是带路径的完整接口地址；确认 Key 没有多余空格。如果用的是.env文件，注意 Django 不会自动加载，需要python-dotenv或手动export。
第二个报错是local proxy failed或连接超时。这类问题通常出在网络层，检查你的运行环境是否能正常访问外网 API，公司内网可能需要配置出口白名单。注意不要使用任何非正规的网络工具，合规访问即可。如果本地能通、线上不通，优先看线上容器的 DNS 和出网策略。
第三个报错是reading choices相关的解析失败，表现为AttributeError: 'NoneType' object has no attribute 'choices'或响应体里没有choices字段。这通常是模型 ID 写错，或者请求体格式不对。确认model字段用的是模型列表里的准确 ID，messages是标准数组格式。如果返回的是错误 JSON，先打印resp原始内容再解析。
第四个是 OAuth 相关报错，比如OAuth token expired或invalid_grant。如果你用的是需要 OAuth 的客户端（比如某些 CLI 工具），token 过期后要重新授权。在 Django 项目里更推荐直接用 API Key 方式，避免 OAuth 刷新逻辑侵入业务代码。
还有一个容易忽略的点：cursor.execute在事务里执行后没提交，读到的可能是旧数据。Django 默认 autocommit 模式下没问题，但如果你手动开了transaction.atomic()，记得在合适位置提交或让上下文管理器处理。
排查时建议打开 Django 的 SQL 日志，在settings.py里加：
LOGGING = { "version": 1, "disable_existing_loggers": False, "handlers": {"console": {"class": "logging.StreamHandler"}}, "loggers": { "django.db.backends": { "handlers": ["console"], "level": "DEBUG", }, }, }这样每条 SQL 和参数都会打出来，参数化是否生效一目了然。注意线上别开 DEBUG 级别，日志量会很大。
6. 把统一 Key 接进你的 Django 调试链路
原生 SQL 的参数化本身不复杂，难的是把它放进一个可验证、可复现的调试链路里。我的做法是：数据库层用cursor()和raw()的参数化写法保证安全，验证层用EXPLAIN和特殊字符输入做回归，模型层用 TaoToken 统一 Key 做 SQL 审查和慢查询解释。三件事各司其职，互不干扰。
如果你还没配 Key，从 API Keys 页面开始：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 。接入细节和参数说明看文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。想先在网页里试模型效果，用模型对话：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。长期做编码和 Agent 任务，看 Coding Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。
最后留一个实用技巧：把常用的参数化查询封装成项目内的db_utils.py，统一处理游标关闭、字典转换、慢查询日志，业务视图里只传 SQL 和参数。这样既保留了原生 SQL 的灵活性，又不会让安全细节散落在各个视图里。下次再遇到IN查询或者动态排序字段，先想清楚哪些能参数化、哪些必须白名单校验，别图省事拼字符串。
