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_ascii=False) cursor.execute("select * from test") print(fetch_dict_result(cursor))

ensure_ascii=False是关键，不加的话中文会变成\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", errors="replace") raise TypeError(f"Type {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_ascii=False, default=json_default)

这样datetime会输出成"2024-01-01 10:00:00"，Decimal变成浮点数，bytes自动解码。注意Decimal转float在金额场景可能有精度损失，如果对精度敏感，改成str(obj)更稳妥。

2.3 用 DictCursor 一步到位

如果你不想手动 zip，MySQLdb 和 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_ascii=False, default=json_default))

DictCursor的代价是内存占用略高，因为每行都建了字典。数据量小、追求代码简洁就用它；数据量大、要流式处理，还是用普通游标加手动映射更可控。

3. TaoToken 统一 Key 前置配置

脚本写好了，怎么验证它真的能跑通？我习惯用一个统一的模型调用通道来生成测试数据、校验 JSON 结构，避免每次都手动造数据。TaoToken 提供的就是这样一个统一 Key 通道，一个 Key 可以走多个模型，省去分别申请和切换的麻烦。

3.1 获取 API Key

先到控制台创建 Key，地址是：

https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite

创建后复制那串sk-开头的 Key，只显示一次，丢了只能重建。建议直接写进环境变量，别硬编码进脚本：

export TAOTOKEN_API_KEY="sk-你的key"

3.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": f"Bearer {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_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite

3.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", errors="replace") raise TypeError(f"Type {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_ascii=False, default=json_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": f"Bearer {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, headers=headers, json=payload, timeout=30) 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和charset，新手最容易忽略。

5. 验证请求与成功结果

配置写完后，跑一次完整流程，确认每一步的输出都符合预期。

5.1 运行脚本

python convert_demo.py

5.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 中文变成 \uXXXX

ensure_ascii默认是True，改成False即可。如果改了还是乱码，检查数据库连接有没有设charset="utf8mb4"，以及表本身的字符集。

6.3 cursor.description 为 None

说明execute还没执行，或者执行的是commit之类的非查询语句。description只在查询后有值，顺序不能颠倒。

6.4 字段名重复导致数据被覆盖

如果select a.id, b.id from a join b，两个列名都叫id，dict(zip(...))后面的会覆盖前面的。解决办法是在 SQL 里用别名：select a.id as a_id, b.id as b_id。

6.5 请求返回 401 或 403

Key 没读到或写错了。先确认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_source=taotoken_aicg_blog_end&utm_content=doc&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/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite

Key 管理和新建入口在控制台：

https://taotoken.net/console/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite

最后提醒一句：fetch_dict_result这个函数建议单独抽成一个工具模块，别在每个查询里重复写。字段映射、类型处理、ensure_ascii这些细节统一在一处维护，后面加字段或改时间格式时只改一个地方，比到处复制粘贴靠谱得多。