python查询MySQL结果输出dict/json：TaoToken统一Key通道下的字段映射与序列化实践

1. Python 查询 MySQL 输出 dict/json 的完整链路与常见坑把 MySQL 查询结果直接变成接口能返回的 JSON是后端和数据处理脚本里出现频率极高的需求。核心检索词就是 python 查询 MySQL 输出 dict/json它要做的事情其实就三件连上数据库、把行数据变成字典、把字典序列化成 JSON 字符串。听起来简单但真正落地时会遇到字段类型全变字符串、中文乱码、datetime 无法序列化、嵌套结构丢失等一堆问题。适合谁看写 Flask/FastAPI 接口的、做数据同步脚本的、用 Python 跑报表导出的以及刚接触 mysql-connector 想搞清楚 cursor(dictionaryTrue) 到底改了什么的同学。我试过最原始的写法用默认游标 fetchall()拿到的是元组列表[(1,2), (11,22)]然后自己拿 cursor.description 去拼字段名再 zip 成字典。代码又长又容易错字段顺序一变就全乱。后来发现 mysql-connector-python 的 cursor 本身就支持 dictionary 参数一行就能拿到[{one:1,tow:2}]。但新的坑来了——所有值都是字符串数字1变成1前端拿到后做加法直接字符串拼接。再往后接 json.dumpsdatetime 和 Decimal 又抛 TypeError。这篇就按真实链路走一遍先讲清楚问题场景和字段映射的坑再说明为什么需要一条稳定的模型/接口通道来辅助调试比如用 TaoToken 统一 Key 通道跑通模型侧的数据处理逻辑然后给出可复制的连接配置和 cursor 字典转换代码接着用一条验证命令确认查询到 JSON 输出成功再对照真实报错做排查最后给出语义一致的入口。全程代码可跟做参数可复制。先明确一个概念MySQL 驱动返回的「字典化」和「JSON 序列化」是两件事。cursor(dictionaryTrue) 只负责把每一行变成 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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 地址https://taotoken.net/api模型对话验证模型是否通https://taotoken.net/api/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewriteCoding Plan长期编码/Agent 场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewriteAPI Keys 管理https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewriteClaude Code Anthropic 接入https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite如果你用的是 Cline、CC Switch 或者 Codex 这类工具配置时三件套要写全Base URL 填https://taotoken.net/apiKey 填你在 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, paramsNone): conn mysql.connector.connect(**DB_CONFIG) try: cur conn.cursor(dictionaryTrue) 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, errorsreplace) raise TypeError(fObject of type {type(obj)} is not JSON serializable) def to_json(rows, ensure_asciiFalse, indent2): return json.dumps(rows, defaultjson_default, ensure_asciiensure_ascii, indentindent)ensure_asciiFalse是关键不然中文会变成\u4e2d\u6587。indent2方便调试生产环境可以去掉省带宽。如果你需要嵌套结构比如把订单和明细拼成树可以在 SQL 层用 JSON_OBJECT 和 JSON_ARRAYAGGMySQL 5.7select 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 字段是字符串形式的 JSONPython 侧再json.loads一次就变成真正的嵌套结构。注意 json_arrayagg 返回的是字符串不是 Python 对象别直接当 list 用。参数对照表参数作用推荐值dictionary行转 dictTruebuffered一次性拉取True小结果集raw返回 bytearrayFalsenamed_tuple返回具名元组按需charset字符集utf8mb4use_unicode返回 strTrue4. 验证请求一条命令确认查询到 JSON 输出成功写完代码要验证。最直接的方式是写一个端到端脚本从查询到 JSON 字符串打印出来。保存为check_json.pyimport 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(fnot serializable: {type(obj)}) def main(): conn mysql.connector.connect(**DB_CONFIG) cur conn.cursor(dictionaryTrue) cur.execute(select 1 as one, 2 as tow union select 11, 22;) rows cur.fetchall() cur.close() conn.close() text json.dumps(rows, defaultjson_default, ensure_asciiFalse) 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 UnauthorizedKey 不对或没带。检查 Authorization 头是不是Bearer keyKey 有没有过期。去 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 能出 dictprint 一行再确认 json.dumps 不报错最后才查模型通道。分层排查比一把梭快得多。6. 语义一致 CTA按场景选入口链路跑通后按你的实际场景选下一步入口别只收藏首页。如果你卡在接入和排障比如 401、Base URL 写错、Key 不会生成直接去 API Keys 管理页拿 Key再对照接入文档逐项核对https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 和 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite如果你只是想验证某个模型能不能通、返回格式对不对用模型对话页面发一条最小请求最快https://taotoken.net/api/chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite如果你是长期跑编码任务、Agent 或者数据同步脚本需要稳定额度看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite控制台入口在这里方便统一管理https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite最后给一个实用技巧把 DB_CONFIG 和模型配置都放环境变量脚本里用 os.environ 读。这样本地调试和线上部署用同一份代码只换环境变量。数据库密码和模型 Key 都不进代码库安全又省心。查询到 JSON 这条链路本身不复杂复杂的是类型和编码的边界情况把 json_default 和 charset 这两处守住基本就稳了。