Python 读取 Mysqldb cursor.fetchall() 结果转 JSON：TaoToken 统一 Key 配置与验证

1. 为什么 cursor.fetchall() 的结果不能直接 json.dumps如果你用 Python 的 MySQLdb或它的现代替代品 pymysql、mysqlclient查过数据库大概率写过这样的代码cursor.execute(select id, name, created_at from users limit 3) rows cursor.fetchall() print(rows)打印出来是((1, alice, datetime.datetime(2024, 1, 1, 10, 0)), (2, bob, ...))这样的元组套元组。这时候你想把它丢给前端接口或者写进日志文件顺手来一句json.dumps(rows)程序立刻给你脸色看TypeError: Object of type datetime is not JSON serializable就算字段全是字符串和数字json.dumps也只会把元组转成 JSON 数组字段名全丢了前端拿到[[1,alice],[2,bob]]根本不知道哪列是哪列。这就是本篇要解决的核心问题cursor.fetchall() 返回的是「无列名的行数据」而 JSON 需要的是「带键值对的对象」。适合读这篇的人有三类正在写数据接口、需要把查询结果直接返回 JSON 的后端同学要把数据库快照落盘成日志、方便后续排查的运维或数据同学以及刚开始用 Python 连 MySQL、被TypeError卡住的初学者。下面我会先讲清楚元组列表转字典的原理再给出可直接复制的转换脚本最后用 TaoToken 的统一 Key 通道跑一次真实调用验证确保你拿到的脚本不是纸上谈兵。2. 转换的核心从 cursor.description 拿到列名很多人卡住是因为只盯着fetchall()的返回值忽略了游标上另一个关键属性cursor.description。执行完execute之后cursor.description会返回一个元组序列每个元素描述一列其中第 0 个位置就是列名。cursor.execute(select id, name, created_at from users limit 1) print(cursor.description) # ((id, 3, None, ...), (name, 253, None, ...), (created_at, 7, None, ...))所以思路很清晰列名列表 每行数据 → zip 配对 → dict → 整个列表 → json.dumps。这就是最经典的fetch_dict_result写法。但直接照搬网上那段代码你会遇到两个坑一是datetime、Decimal这类类型json.dumps不认二是中文默认被转成\uXXXX转义日志里没法看。2.1 基础版转换函数先给一个能跑的最小骨架字段映射逻辑一目了然import json def fetch_dict_result(cur): # 从 description 提取列名 row_headers [x[0] for x in cur.description] rv cur.fetchall() json_data [] for result in rv: json_data.append(dict(zip(row_headers, result))) return json.dumps(json_data, ensure_asciiFalse) cursor.execute(select * from test) print(fetch_dict_result(cursor))ensure_asciiFalse是关键不加的话中文会变成\u4f60\u597d这种落盘日志基本没法读。2.2 处理 datetime 和 Decimal真实表里几乎一定有datetime、date、Decimal字段json.dumps默认不认识它们。最省事的做法是自定义default回调import json from datetime import date, datetime 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): return obj.decode(utf-8, errorsreplace) raise TypeError(fType {type(obj)} not serializable) def fetch_dict_result(cur): row_headers [x[0] for x in cur.description] rv cur.fetchall() json_data [dict(zip(row_headers, row)) for row in rv] return json.dumps(json_data, ensure_asciiFalse, defaultjson_default)这样datetime会输出成2024-01-01 10:00:00Decimal变成浮点数bytes自动解码。注意Decimal转float在金额场景可能有精度损失如果对精度敏感改成str(obj)更稳妥。2.3 用 DictCursor 一步到位如果你不想手动 zipMySQLdb 和 pymysql 都支持DictCursor查询结果直接就是字典import MySQLdb.cursors cursor conn.cursor(MySQLdb.cursors.DictCursor) cursor.execute(select id, name from users limit 3) rows cursor.fetchall() # 直接是 [{id:1,name:alice}, ...] print(json.dumps(rows, ensure_asciiFalse, defaultjson_default))DictCursor的代价是内存占用略高因为每行都建了字典。数据量小、追求代码简洁就用它数据量大、要流式处理还是用普通游标加手动映射更可控。3. TaoToken 统一 Key 前置配置脚本写好了怎么验证它真的能跑通我习惯用一个统一的模型调用通道来生成测试数据、校验 JSON 结构避免每次都手动造数据。TaoToken 提供的就是这样一个统一 Key 通道一个 Key 可以走多个模型省去分别申请和切换的麻烦。3.1 获取 API Key先到控制台创建 Key地址是https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite创建后复制那串sk-开头的 Key只显示一次丢了只能重建。建议直接写进环境变量别硬编码进脚本export TAOTOKEN_API_KEYsk-你的key3.2 接口地址与鉴权方式TaoToken 的 API 基地址是https://taotoken.net/api兼容 OpenAI 风格的/v1/chat/completions。鉴权走标准的Authorization: Bearer头。Python 里用requests调用的最小配置如下import os import requests API_KEY os.environ[TAOTOKEN_API_KEY] BASE_URL https://taotoken.net/api/v1/chat/completions headers { Authorization: fBearer {API_KEY}, Content-Type: application/json, } payload { model: gpt-4o-mini, messages: [ {role: user, content: 生成一条包含 id、name、created_at 的 JSON 示例} ], }模型名按你账号里可用的填具体清单可以在模型对话页确认https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite3.3 为什么验证环节要用它单纯测fetch_dict_result其实不需要联网但真实项目里 JSON 结构往往要和接口约定对齐比如字段名是created_at还是createTime、时间格式是字符串还是时间戳。用统一 Key 通道让模型帮你生成一份「期望的 JSON 样例」再拿你的脚本输出去比对比人肉核对快得多。这就是前置配置的意义把「转换脚本」和「结构校验」串成一条可复现的链路。4. 可复制配置完整转换与验证脚本下面这份脚本把数据库查询、元组转 JSON、以及通过 TaoToken 做结构校验串在一起可以直接改连接参数后运行。4.1 数据库连接与查询部分import json import os from datetime import date, datetime from decimal import Decimal import MySQLdb import MySQLdb.cursors import requests DB_CONFIG { host: 127.0.0.1, port: 3306, user: root, passwd: your_password, db: test_db, charset: utf8mb4, } def json_default(obj): if isinstance(obj, (datetime, date)): return obj.strftime(%Y-%m-%d %H:%M:%S) if isinstance(obj, Decimal): return str(obj) if isinstance(obj, bytes): return obj.decode(utf-8, errorsreplace) raise TypeError(fType {type(obj)} not serializable) def fetch_dict_result(cur): row_headers [x[0] for x in cur.description] rv cur.fetchall() json_data [dict(zip(row_headers, row)) for row in rv] return json.dumps(json_data, ensure_asciiFalse, defaultjson_default) def query_users(): conn MySQLdb.connect(**DB_CONFIG) try: cur conn.cursor() cur.execute(select id, name, created_at from users limit 5) return fetch_dict_result(cur) finally: conn.close()注意finally里关连接别让异常路径漏掉close。charset一定要设utf8mb4否则中文和 emoji 会乱码。4.2 通过 TaoToken 校验 JSON 结构拿到查询结果后把 JSON 丢给模型让它检查字段是否齐全、类型是否合理def validate_json_structure(json_str): api_key os.environ[TAOTOKEN_API_KEY] url https://taotoken.net/api/v1/chat/completions headers { Authorization: fBearer {api_key}, Content-Type: application/json, } prompt ( 下面是一段数据库查询导出的 JSON请检查\n 1. 是否每个对象都有 id、name、created_at 三个字段\n 2. created_at 是否为可读的时间字符串\n 3. 指出任何类型异常。只返回问题列表没问题就返回 OK。\n\n f{json_str} ) payload { model: gpt-4o-mini, messages: [{role: user, content: prompt}], temperature: 0, } resp requests.post(url, headersheaders, jsonpayload, timeout30) resp.raise_for_status() return resp.json()[choices][0][message][content] if __name__ __main__: result query_users() print(查询结果, result) print(结构校验, validate_json_structure(result))temperature设 0 是为了让校验结果稳定别让它自由发挥。timeout一定要加网络请求没有超时是生产事故的常见来源。4.3 关键参数对照参数作用建议值ensure_ascii中文是否转义Falsedefault非标准类型处理自定义json_defaultcharset数据库字符集utf8mb4temperature校验稳定性0timeout请求超时30秒这张表里的每一项都是踩过坑总结出来的尤其是ensure_ascii和charset新手最容易忽略。5. 验证请求与成功结果配置写完后跑一次完整流程确认每一步的输出都符合预期。5.1 运行脚本python convert_demo.py5.2 预期输出数据库部分应该输出类似这样的 JSON 字符串[{id: 1, name: alice, created_at: 2024-01-01 10:00:00}, {id: 2, name: bob, created_at: 2024-01-02 11:30:00}]注意created_at已经是可读字符串不是datetime对象中文也没有被转义。校验部分应该返回OK或者列出具体问题比如「第 2 个对象缺少 created_at 字段」。如果返回 OK说明你的转换脚本和字段约定完全对齐。5.3 用 curl 单独验证 Key 通道如果你想先确认 Key 本身没问题可以脱离 Python 直接用 curl 测一次curl https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer $TAOTOKEN_API_KEY \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: 回复 OK}] }返回体里choices[0].message.content是OK就说明 Key 和网络都正常。这一步能帮你快速区分「是脚本问题」还是「是 Key 问题」。6. 本篇常见错误排查转换脚本本身不复杂但报错信息往往指向不同层面下面按出现频率排一下。6.1 TypeError: Object of type datetime is not JSON serializable最常见原因是没传default回调。解决方式是加上json_default并在json.dumps里引用。如果你用的是DictCursor同样要传default因为字典里的值还是datetime对象。6.2 中文变成 \uXXXXensure_ascii默认是True改成False即可。如果改了还是乱码检查数据库连接有没有设charsetutf8mb4以及表本身的字符集。6.3 cursor.description 为 None说明execute还没执行或者执行的是commit之类的非查询语句。description只在查询后有值顺序不能颠倒。6.4 字段名重复导致数据被覆盖如果select a.id, b.id from a join b两个列名都叫iddict(zip(...))后面的会覆盖前面的。解决办法是在 SQL 里用别名select a.id as a_id, b.id as b_id。6.5 请求返回 401 或 403Key 没读到或写错了。先确认echo $TAOTOKEN_API_KEY有值再确认请求头是Bearer加空格加 Key。如果 Key 是在别的终端创建的当前终端没export也会读不到。6.6 请求超时网络抖动或模型响应慢。给requests.post加timeout并在外层做一次重试。重试次数别太多两三次足够避免雪崩。6.7 Decimal 转 float 精度丢失金额字段别用float(obj)改成str(obj)让前端或下游自己决定怎么解析。这是财务类项目必须注意的点。7. 接入文档与后续调用入口脚本跑通之后如果你要把这套转换逻辑接到真实的数据接口或日志管道里建议先过一遍接入文档确认鉴权头、错误码和限流规则https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite需要长期跑编码任务、或者把这类校验逻辑做成 Agent 定时执行的可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite只想快速对话验证模型输出的直接进模型对话页https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewriteKey 管理和新建入口在控制台https://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite最后提醒一句fetch_dict_result这个函数建议单独抽成一个工具模块别在每个查询里重复写。字段映射、类型处理、ensure_ascii这些细节统一在一处维护后面加字段或改时间格式时只改一个地方比到处复制粘贴靠谱得多。