1. Python 查询 MySQL 输出 dict/json 的完整链路与常见坑
把 MySQL 查询结果直接变成接口能返回的 JSON,是后端和数据处理脚本里出现频率极高的需求。核心检索词就是 python 查询 MySQL 输出 dict/json,它要做的事情其实就三件:连上数据库、把行数据变成字典、把字典序列化成 JSON 字符串。听起来简单,但真正落地时会遇到字段类型全变字符串、中文乱码、datetime 无法序列化、嵌套结构丢失等一堆问题。适合谁看?写 Flask/FastAPI 接口的、做数据同步脚本的、用 Python 跑报表导出的,以及刚接触 mysql-connector 想搞清楚 cursor(dictionary=True) 到底改了什么的同学。
我试过最原始的写法:用默认游标 fetchall(),拿到的是元组列表[('1','2'), ('11','22')],然后自己拿 cursor.description 去拼字段名,再 zip 成字典。代码又长又容易错,字段顺序一变就全乱。后来发现 mysql-connector-python 的 cursor 本身就支持 dictionary 参数,一行就能拿到[{'one':'1','tow':'2'}]。但新的坑来了——所有值都是字符串,数字1变成'1',前端拿到后做加法直接字符串拼接。再往后接 json.dumps,datetime 和 Decimal 又抛 TypeError。
这篇就按真实链路走一遍:先讲清楚问题场景和字段映射的坑,再说明为什么需要一条稳定的模型/接口通道来辅助调试(比如用 TaoToken 统一 Key 通道跑通模型侧的数据处理逻辑),然后给出可复制的连接配置和 cursor 字典转换代码,接着用一条验证命令确认查询到 JSON 输出成功,再对照真实报错做排查,最后给出语义一致的入口。全程代码可跟做,参数可复制。
先明确一个概念:MySQL 驱动返回的「字典化」和「JSON 序列化」是两件事。cursor(dictionary=True) 只负责把每一行变成 Python dict,键是列名,值是驱动按列类型转换后的 Python 对象。而 json.dumps 负责把 Python 对象变成字符串。中间如果类型不兼容,就会在序列化这一步炸掉。所以字段映射要分两层看:驱动层(MySQL 类型 → Python 类型)和序列化层(Python 类型 → JSON 类型)。很多人只做了第一层,第二层直接崩。
还有一个容易被忽略的点:列名重复。比如select a.id, b.id from a join b,字典化后后面的 id 会覆盖前面的,数据静默丢失。这种问题不会报错,但结果就是错的。解决办法是给列起别名,select a.id as a_id, b.id as b_id。这个坑我在做多表关联导出时踩过,排查了半天才发现是键冲突。
2. TaoToken 统一 Key 通道:为数据处理链路提供稳定调用入口
在把查询结果转 JSON 的过程中,经常需要配合模型做字段清洗、结构补全或者接口联调。比如你查出来的 JSON 要喂给一个模型做摘要,或者用模型帮你把不规则的字段名映射成统一 schema。这时候如果每个模型都单独配 Key、单独改 Base URL,调试成本会很高。TaoToken 提供的是统一 Key 通道,一个 Key 走多个模型,Base URL 固定,适合这种「查询 → 处理 → 输出」的链路。
需要说清楚的是,TaoToken 在这里的角色是模型调用的统一入口,不是数据库代理,也不碰你的 MySQL 连接。你的 Python 脚本依然用 mysql-connector 直连数据库,只是在需要调用模型做后处理时,把请求发到统一通道。这样数据库配置和模型配置解耦,换模型不用动数据库代码。
接入信息如下,建议直接复制:
- 官网入口:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
- API 地址:https://taotoken.net/api
- 模型对话(验证模型是否通):https://taotoken.net/api/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite
- Coding Plan(长期编码/Agent 场景):https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
- 控制台:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite
- 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
- Claude Code Anthropic 接入:https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude_code&utm_campaign=rewrite
如果你用的是 Cline、CC Switch 或者 Codex 这类工具,配置时三件套要写全:Base URL 填https://taotoken.net/api,Key 填你在 API Keys 页面生成的令牌,Model ID 填你要用的模型名。缺一个都会连不上。我见过有人只填了 Key 没改 Base URL,结果一直报 401,排查半天。
为什么数据处理场景需要这个?举个例子:你从 MySQL 查出一批订单,字段是order_id, amount, created_at,但下游接口要求id, total, timestamp。你可以写死映射,也可以用模型做动态映射。用统一通道的好处是,映射逻辑的模型调用和数据库查询在同一个脚本里,Key 只维护一份。对于长期跑的同步任务,Coding Plan 更合适,不用每次担心额度。
需要提醒:TaoToken 是模型调用的统一入口,不替代你的数据库连接,也不替代编辑器。数据库该用 mysql-connector 还是用它,模型调用才走这个通道。两者职责分开,脚本才清晰。
3. 可复制配置:连接参数、cursor 字典转换与 json.dumps 模板
这一节是核心,直接给可复制的代码。先装依赖:
pip install mysql-connector-python连接配置建议单独放一个 dict,方便复用。注意 charset 要写utf8mb4,不然中文和 emoji 会出问题:
import mysql.connector from mysql.connector import Error DB_CONFIG = { "host": "127.0.0.1", "port": 3306, "user": "your_user", "password": "your_password", "database": "your_db", "charset": "utf8mb4", "use_unicode": True, "autocommit": True, }cursor 字典化有两种写法。第一种是连接时指定,第二种是创建 cursor 时指定。推荐第二种,灵活:
def query_as_dict(sql, params=None): conn = mysql.connector.connect(**DB_CONFIG) try: cur = conn.cursor(dictionary=True) cur.execute(sql, params or ()) rows = cur.fetchall() return rows finally: cur.close() conn.close()调用:
sql = """ select 1 as one, 2 as tow union select 11 as one, 22 as tow union select '011' as one, '022' as tow; """ result = query_as_dict(sql) print(result)输出是[{'one': '1', 'tow': '2'}, {'one': '11', 'tow': '22'}, {'one': '011', 'tow': '022'}]。注意这里值全是字符串,因为 union 里混了字符串字面量,MySQL 会把整列当字符串处理。这是字段映射的第一个坑:列类型由查询结果决定,不由你想象决定。
接下来是 json.dumps 模板。直接 dumps 会碰到 datetime、Decimal、bytes 报错。写一个 default 处理器:
import json from datetime import datetime, date from decimal import Decimal def json_default(obj): if isinstance(obj, (datetime, date)): return obj.strftime("%Y-%m-%d %H:%M:%S") if isinstance(obj, Decimal): return float(obj) if isinstance(obj, (bytes, bytearray)): return obj.decode("utf-8", errors="replace") raise TypeError(f"Object of type {type(obj)} is not JSON serializable") def to_json(rows, ensure_ascii=False, indent=2): return json.dumps(rows, default=json_default, ensure_ascii=ensure_ascii, indent=indent)ensure_ascii=False是关键,不然中文会变成\u4e2d\u6587。indent=2方便调试,生产环境可以去掉省带宽。
如果你需要嵌套结构,比如把订单和明细拼成树,可以在 SQL 层用 JSON_OBJECT 和 JSON_ARRAYAGG(MySQL 5.7+):
select o.id as order_id, o.amount, json_arrayagg(json_object('sku', d.sku, 'qty', d.qty)) as items from orders o join order_detail d on d.order_id = o.id group by o.id, o.amount;这样查出来的 items 字段是字符串形式的 JSON,Python 侧再json.loads一次就变成真正的嵌套结构。注意 json_arrayagg 返回的是字符串,不是 Python 对象,别直接当 list 用。
参数对照表:
| 参数 | 作用 | 推荐值 |
|---|---|---|
| dictionary | 行转 dict | True |
| buffered | 一次性拉取 | True(小结果集) |
| raw | 返回 bytearray | False |
| named_tuple | 返回具名元组 | 按需 |
| charset | 字符集 | utf8mb4 |
| use_unicode | 返回 str | True |
4. 验证请求:一条命令确认查询到 JSON 输出成功
写完代码要验证。最直接的方式是写一个端到端脚本,从查询到 JSON 字符串打印出来。保存为check_json.py:
import json import mysql.connector from datetime import datetime from decimal import Decimal DB_CONFIG = { "host": "127.0.0.1", "port": 3306, "user": "your_user", "password": "your_password", "database": "your_db", "charset": "utf8mb4", "use_unicode": True, } def json_default(obj): if isinstance(obj, (datetime,)): return obj.strftime("%Y-%m-%d %H:%M:%S") if isinstance(obj, Decimal): return float(obj) raise TypeError(f"not serializable: {type(obj)}") def main(): conn = mysql.connector.connect(**DB_CONFIG) cur = conn.cursor(dictionary=True) cur.execute("select 1 as one, 2 as tow union select 11, 22;") rows = cur.fetchall() cur.close() conn.close() text = json.dumps(rows, default=json_default, ensure_ascii=False) print(text) assert isinstance(rows, list) and isinstance(rows[0], dict) assert json.loads(text) == rows print("OK: query -> dict -> json verified") if __name__ == "__main__": main()运行:
python check_json.py预期输出:
[{"one": "1", "tow": "2"}, {"one": "11", "tow": "22"}] OK: query -> dict -> json verified看到OK就说明链路通了。这里做了两个断言:rows 是 list 且元素是 dict,以及 json.loads 后和原数据一致。第二个断言很重要,它能发现序列化过程中的类型丢失。比如 Decimal 转 float 后精度可能变,断言会帮你发现。
如果你要验证模型侧通道是否通,可以用 curl 发一条最小请求:
curl -s https://taotoken.net/api/chat \ -H "Authorization: Bearer $TAOTOKEN_KEY" \ -H "Content-Type: application/json" \ -d '{"model":"your-model-id","messages":[{"role":"user","content":"ping"}]}'返回里有 choices 字段就说明通道正常。注意 Base URL 是https://taotoken.net/api,不要多加斜杠或路径。Key 从 API Keys 页面拿,别硬编码在脚本里,用环境变量。
验证通过后,把query_as_dict和to_json封装成模块,接口层直接return to_json(query_as_dict(sql))就行。FastAPI 里可以直接返回 dict,框架会帮你序列化,但 datetime 还是要自己处理,所以统一走 to_json 更稳。
5. 本篇常见错排查:401、local proxy failed、reading choices、OAuth
这一节对照真实报错。数据库侧和模型侧的问题分开看。
数据库侧最常见的三个:
第一个是中文乱码,表现为查出来是????或\xe4\xb8\xad。原因是连接没指定 charset。解决:DB_CONFIG 里加"charset": "utf8mb4"和"use_unicode": True。如果表本身是 latin1,还要改表字符集。
第二个是TypeError: Object of type datetime is not JSON serializable。这是 json.dumps 没给 default。解决:加上 json_default 处理器,把 datetime 转字符串,Decimal 转 float。
第三个是字段名重复导致数据丢失。表现是 join 查询后某个字段值不对。解决:SQL 里给同名列起别名,a.id as a_id。
模型侧通道的报错:
401 Unauthorized:Key 不对或没带。检查 Authorization 头是不是Bearer <key>,Key 有没有过期。去 API Keys 页面重新生成一个。注意别把 Key 提交到 git。
local proxy failed:本地网络配置问题,请求没发出去。检查 Base URL 是不是写成了https://taotoken.net/api/(末尾多斜杠),或者环境变量里有没有残留的代理设置。把 Base URL 改成https://taotoken.net/api再试。
reading choices相关报错:通常是响应体不是预期 JSON,比如返回了 HTML 错误页。用 curl 加-i看状态码和 Content-Type。如果是 502/503,稍后重试;如果是 404,检查路径。
OAuth相关报错:多见于 Claude Code 或 Codex 这类工具的接入。这类工具要求三件套齐全:Base URL、Key、Model ID。缺 Model ID 会报模型不存在,缺 Base URL 会走默认官方地址导致鉴权失败。配置时逐项核对,别只填一半。
还有一个隐蔽的:mysql.connector.errors.OperationalError: 2013 Lost connection。长查询或网络抖动会触发。解决:加connection_timeout和autocommit,或者用连接池。小脚本可以加重试。
排查顺序建议:先确认数据库能连(用 mysql 命令行试),再确认 cursor 能出 dict(print 一行),再确认 json.dumps 不报错,最后才查模型通道。分层排查比一把梭快得多。
6. 语义一致 CTA:按场景选入口
链路跑通后,按你的实际场景选下一步入口,别只收藏首页。
如果你卡在接入和排障,比如 401、Base URL 写错、Key 不会生成,直接去 API Keys 管理页拿 Key,再对照接入文档逐项核对: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/api/chat?utm_source=taotoken_aicg_blog_end&utm_content=model_chat&utm_campaign=rewrite
如果你是长期跑编码任务、Agent 或者数据同步脚本,需要稳定额度,看 Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite
控制台入口在这里,方便统一管理:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite
最后给一个实用技巧:把 DB_CONFIG 和模型配置都放环境变量,脚本里用 os.environ 读。这样本地调试和线上部署用同一份代码,只换环境变量。数据库密码和模型 Key 都不进代码库,安全又省心。查询到 JSON 这条链路本身不复杂,复杂的是类型和编码的边界情况,把 json_default 和 charset 这两处守住,基本就稳了。