☰
Django5.1(23)—— 执行原生 SQL 查询:用 TaoToken 统一 Key 打通多模型 SQL 生成与校验
2026/10/2 6:07:30 网站建设 项目流程

1. 为什么 ORM 写不动了:一个真实的多表聚合场景

先说结论:Django ORM 能覆盖 90% 的日常查询,但剩下 10% 的复杂聚合、窗口函数、递归 CTE,硬用 ORM 拼出来的代码往往比原生 SQL 还难维护。我最近接手一个报表模块,需求是按「用户 + 月份 + 订单状态」做多维汇总,还要算环比和累计值。用annotate+Subquery拼了两百多行,跑起来还慢,最后老老实实回到原生 SQL。

这个场景的典型特征是:分组维度动态、需要窗口函数(ROW_NUMBER、SUM() OVER)、还要跨表关联三四张表。ORM 的Window表达式虽然支持一部分,但一旦涉及动态列名和复杂CASE WHEN,可读性断崖式下跌。这时候Manager.raw()和connection.cursor()就是正解。

但原生 SQL 有两个绕不开的坑:一是 SQL 注入,二是 SQL 本身写错。前者靠参数化占位符解决,后者靠校验。我试过用 TaoToken 的统一 Key 调多个模型来生成 SQL 并交叉校验,再回填到 Django 执行,整个链路顺下来比纯手写稳不少。这篇就把这套流程拆开讲清楚,包括 settings 配置、raw 查询、参数化、EXPLAIN 验证,以及几个我踩过的报错。

适合谁看:已经会写 Django 模型和 QuerySet,但遇到复杂查询卡住的开发者;或者想给团队引入「AI 辅助生成 SQL + 人工校验」流程的人。核心检索词就是 Django 原生 SQL 查询、raw 执行、参数化防注入,下面逐步展开。

2. 前置准备:用 TaoToken 统一 Key 打通多模型 SQL 生成

在写 SQL 之前,先把「生成 + 校验」这条链路搭起来。思路很简单:Django 项目里遇到 ORM 表达不了的查询,先把表结构和需求描述丢给模型,让它产出候选 SQL,再用另一个模型做 review,最后人工确认后回填。TaoToken 在这里的作用是提供一个统一的 API 入口,一个 Key 就能切换不同模型,不用为每个模型单独配环境变量。

先拿 Key。打开 https://taotoken.net/api-keys ,登录后创建一个 API Key,复制保存。注意这个 Key 只在创建时完整显示一次,丢了就得重建。拿到之后,建议放在项目根目录的.env里,别硬编码进settings.py。

# .env TAOTOKEN_API_KEY=sk-你的key TAOTOKEN_BASE_URL=https://taotoken.net/api

然后在settings.py里读取。如果你用django-environ或python-dotenv,直接加载即可:

# settings.py import os from dotenv import load_dotenv load_dotenv() TAOTOKEN_API_KEY = os.getenv("TAOTOKEN_API_KEY") TAOTOKEN_BASE_URL = os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api")

这里要强调一点:TaoToken 是合规的 API 聚合入口,Base URL 固定用https://taotoken.net/api,不要加任何 UTM 参数到 API 地址上,否则部分 SDK 会把它当成非法路径。官网入口是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,需要看文档或控制台从那里进。

模型选择上,生成 SQL 我一般用推理能力强的模型,校验用另一个模型做交叉检查。TaoToken 的模型对话入口在 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=sql_gen&utm_campaign=rewrite ,可以在网页里先试几轮,确认 prompt 效果再写进代码。如果你打算长期在项目里跑这套流程,甚至接进 CI 做 SQL lint,可以考虑 Coding Plan,入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=sql_gen&utm_campaign=rewrite ,按量或包月看团队规模。

配置片段建议单独放一个模块,方便复用:

# utils/ai_sql.py import os from openai import OpenAI client = OpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api"), ) def generate_sql(schema: str, requirement: str, model: str = "gpt-4o") -> str: resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "你是 PostgreSQL 专家,只输出 SQL,不要解释。"}, {"role": "user", "content": f"表结构:\n{schema}\n\n需求:{requirement}"}, ], temperature=0.2, ) return resp.choices[0].message.content.strip()

这段代码里base_url指向 TaoToken,model参数可以换成任意 TaoToken 支持的模型 ID。一个 Key 打通多模型,切换只改一个字符串,这是它最实用的地方。schema 建议直接从manage.py inspectdb或数据库\d命令导出,保证模型看到的是真实字段类型。

3. 可复制配置:settings 片段与 raw 查询参数化示例

配置好 Key 之后,回到 Django 本身。原生 SQL 有两条路:Manager.raw()返回模型实例,connection.cursor()完全绕过模型层。先看 settings 里需要确认的数据库配置,再给两套可复制的查询模板。

settings.py的DATABASES部分,确保OPTIONS里没有奇怪的强制类型转换,尤其是 MySQL 用户:

# settings.py DATABASES = { "default": { "ENGINE": "django.db.backends.postgresql", "NAME": os.getenv("DB_NAME", "mydb"), "USER": os.getenv("DB_USER", "postgres"), "PASSWORD": os.getenv("DB_PASSWORD", ""), "HOST": os.getenv("DB_HOST", "127.0.0.1"), "PORT": os.getenv("DB_PORT", "5432"), "OPTIONS": { "connect_timeout": 10, }, } }

如果你用 SQLite 做本地开发,注意一个坑:SQLite 后端不支持字典参数,raw()的params必须传列表。这个后面排障会细说。

先看Manager.raw()的参数化写法。假设模型是Order,表名shop_order:

# apps/shop/models.py from django.db import models class Order(models.Model): user_id = models.IntegerField() amount = models.DecimalField(max_digits=10, decimal_places=2) status = models.CharField(max_length=20) created_at = models.DateTimeField() class Meta: db_table = "shop_order"

用raw()做参数化查询,占位符统一用%s,参数用列表传:

# 正确:参数化,防注入 status = "paid" min_amount = 100 orders = Order.objects.raw( "SELECT id, user_id, amount, status, created_at " "FROM shop_order WHERE status = %s AND amount >= %s", [status, min_amount], ) for o in orders: print(o.id, o.amount)

字段映射靠列名匹配,顺序无所谓。如果 SQL 里的列名和模型字段名不一致,用AS或translations参数:

name_map = {"uid": "user_id", "amt": "amount"} orders = Order.objects.raw( "SELECT id, uid, amt, status FROM shop_order WHERE status = %s", [status], translations=name_map, )

再看connection.cursor()的写法,适合 UPDATE/INSERT 或不映射模型的聚合查询:

from django.db import connection def monthly_summary(year: int, month: int): sql = """ SELECT user_id, SUM(amount) AS total, COUNT(*) AS cnt FROM shop_order WHERE EXTRACT(YEAR FROM created_at) = %s AND EXTRACT(MONTH FROM created_at) = %s GROUP BY user_id ORDER BY total DESC """ with connection.cursor() as cursor: cursor.execute(sql, [year, month]) columns = [col[0] for col in cursor.description] return [dict(zip(columns, row)) for row in cursor.fetchall()]

这里dictfetchall的逻辑直接内联了,省得再定义工具函数。注意cursor.description在execute之后才有值,别提前取。参数依然用%s,Django 会交给底层驱动转义,不要自己用 f-string 拼。

如果你有多个数据库,用connections["alias"]:

from django.db import connections with connections["report_db"].cursor() as cursor: cursor.execute("SELECT COUNT(*) FROM big_table WHERE dt = %s", [dt]) total = cursor.fetchone()[0]

这两套模板覆盖了大部分场景。生成 SQL 的时候,把上面这些表结构和字段名喂给模型,让它按%s占位符输出,回填时直接可用,不用再改占位符风格。

4. 验证请求:EXPLAIN 与成功结果对照

SQL 写出来不能直接上生产,先验证。验证分两层:语法/执行计划层用EXPLAIN,结果正确性层用少量样本数据对照。

先看EXPLAIN。在 Django shell 里跑:

from django.db import connection sql = """ SELECT user_id, SUM(amount) AS total FROM shop_order WHERE status = %s GROUP BY user_id """ with connection.cursor() as cursor: cursor.execute("EXPLAIN " + sql, ["paid"]) for row in cursor.fetchall(): print(row[0])

PostgreSQL 会输出执行计划,重点看有没有Seq Scan全表扫描、有没有走索引。如果EXPLAIN报语法错误,说明 SQL 本身有问题,回到模型生成的候选里换一个。MySQL 用EXPLAIN同样可以,SQLite 用EXPLAIN QUERY PLAN。

我实测下来,模型生成的 SQL 大概有 20% 会在EXPLAIN阶段暴露问题,常见的是GROUP BY漏字段、JOIN条件写错、窗口函数PARTITION BY用错列。交叉校验的价值就在这里:让第二个模型专门检查「这个 SQL 在 PostgreSQL 下能否执行、有没有语法问题」,比人工肉眼扫快得多。

校验 prompt 可以这样写:

def review_sql(schema: str, sql: str, model: str = "claude-3-5-sonnet") -> str: resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": "你是数据库审查员,检查 SQL 语法、注入风险、性能隐患,输出问题列表。"}, {"role": "user", "content": f"表结构:\n{schema}\n\n待审 SQL:\n{sql}"}, ], temperature=0, ) return resp.choices[0].message.content

两个模型跑完,人工确认后再回填到 Django。回填后跑一次真实查询,对照 ORM 的结果做抽样比对。比如同一个月份,用Order.objects.filter(...).aggregate(Sum("amount"))算一个总数,和原生 SQL 的结果比,一致就说明逻辑没问题。

成功的结果长这样:EXPLAIN输出里出现Index Scan using idx_order_status,查询耗时从 1.2s 降到 80ms;fetchall()返回的 dict 列表字段名和预期一致,数值对得上。到这一步,原生 SQL 才算真正落地。

5. 常见报错排查:401、local proxy failed、reading choices、OAuth

这一节列几个我在接入和运行过程中真实遇到的报错,以及对应的排查动作。

401 Unauthorized。调 TaoToken 接口时最常见。先检查.env里的TAOTOKEN_API_KEY有没有多余空格或换行,load_dotenv()是否在读取前执行。再确认base_url是https://taotoken.net/api,没有拼错路径。如果 Key 是从控制台复制的,确认没有把Bearer前缀也复制进去,SDK 会自动加。401 基本就是 Key 或 Base URL 的问题,跟模型无关。

local proxy failed。这个报错通常出现在请求根本没发出去的时候,比如本地网络策略拦截、或者base_url指向了一个不可达的地址。排查顺序:先用curl https://taotoken.net/api/models -H "Authorization: Bearer $TAOTOKEN_API_KEY"测一下连通性,能返回模型列表说明网络没问题;如果 curl 也失败,检查是不是环境变量没生效。注意不要在任何地方配置非官方的转发地址,Base URL 只认https://taotoken.net/api。

reading choices 报错。典型信息是KeyError: 'choices'或AttributeError: 'NoneType' object has no attribute 'choices'。这说明响应体里没有choices字段,通常是模型 ID 写错了,或者请求被拒。先打印完整响应:

resp = client.chat.completions.create(...) print(resp.model_dump())

如果返回的是错误对象,里面会有error.message,按提示改模型 ID 或参数。另一个可能是temperature或max_tokens超了模型限制,调小再试。

OAuth 相关报错。如果你用的是 Claude Code 或某些 CLI 工具,可能会遇到 OAuth 认证失败。这类工具通常需要配置三件套:Base URL、API Key、Model ID。以 Claude Code 为例,在配置文件里写:

{ "base_url": "https://taotoken.net/api", "api_key": "sk-你的key", "model": "claude-3-5-sonnet" }

三个字段缺一不可,Model ID 要和 TaoToken 文档里列出的完全一致。如果工具提示 OAuth 失败,先确认是不是把 API Key 填到了 OAuth token 的位置,两者不是一回事。Cline 的 MCP 配置同理,Base URL + Key + Model ID 三件套写全,少一个都会报认证错误。

排查完这些,基本能覆盖 90% 的接入问题。剩下的多半是 SQL 本身的语法或逻辑错误,回到EXPLAIN那一步处理。

6. 把这条链路固化进你的 Django 项目

整套流程跑通之后,建议把它固化下来,而不是每次临时拼。我的做法是在项目里建一个sql_workbench管理命令,输入需求描述,自动调 TaoToken 生成 SQL、跑EXPLAIN、输出候选,人工确认后再写进代码。这样既保留了 AI 的效率,又守住了人工审核的安全边界。

几个实用技巧:schema 导出用manage.py inspectdb --database default > schema.py,比手写准;生成 SQL 时把%s占位符要求写进 system prompt,回填零改动;EXPLAIN一定要在真实数据集上跑,空表看不出性能问题;参数化永远用列表或字典传params,绝不用 f-string 拼 SQL。

需要长期在项目里跑这套流程的话,Coding Plan 的入口在 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=sql_gen&utm_campaign=rewrite ,接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=sql_gen&utm_campaign=rewrite ,控制台在 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=sql_gen&utm_campaign=rewrite 。先把 Key 和 Base URL 配好,剩下的就是把这套模板套进你的模型和查询里。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询