1. 为什么 pymysql 查出来的数据不能直接 json.dumps
你写了一个本地脚本,或者用 Cline 这类 AI 工具生成了一个数据接口,从 MySQL 里select出几行数据,想直接json.dumps(results)返回给前端或写进文件,结果报错TypeError: Object of type datetime is not JSON serializable,或者Decimal is not JSON serializable。这不是你代码写错了,而是 pymysql 返回的 Python 对象和 JSON 标准类型之间天然存在一道鸿沟。
pymysql 默认把 MySQL 的字段映射成 Python 原生类型:DATETIME/TIMESTAMP变成datetime.datetime,DECIMAL变成decimal.Decimal,BIGINT在部分场景下变成int但超出 JS 安全整数范围,TINYINT(1)变成bool。这些类型json.dumps一个都不认识。更隐蔽的坑是:cursor.fetchall()返回的是元组列表,字段名只存在于cursor.description里,如果你直接序列化元组,字段名全丢了,前端拿到一堆没有 key 的数组,根本没法用。
还有一个新手经常忽略的点:json.dumps默认ensure_ascii=True,中文会被转义成\u300a这种形式。数据本身没错,但可读性极差,调试时看着头疼,接口返回给前端虽然能解析,但日志里全是乱码一样的转义串。
这篇就围绕「pymysql 查询结果转 JSON 字符串」这一件事,把连接配置、字段名还原、自定义 JSONEncoder、中文处理、以及通过 TaoToken 统一 Key 通道做接口验证的完整链路讲清楚。适合刚入门 Python、正在写本地脚本或 AI 工具生成的数据接口的同学,跟着做就能跑通。
2. TaoToken 统一 Key 通道的前置准备
在讲序列化之前,先说清楚为什么这篇要提 TaoToken。你写的是一个数据接口脚本,不管是本地跑还是给 Cline 这类 AI 编码工具调用,最终都要有一个稳定的模型通道来做验证、调试或者生成代码。TaoToken 提供的是统一 Key 通道,一个 Key 走通模型对话、Coding Plan、API 调用,不用在多个平台之间来回切换配置。
你需要先拿到 Key。打开官网 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,注册后在控制台里创建 API Key。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API Keys 管理页在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。Key 拿到后先存好,后面配置settings.json要用。
这里要区分两个地址:官网带 UTM 参数用于追踪来源,API 端点统一用 https://taotoken.net/api ,不加任何 UTM。你的脚本里请求模型接口时,base_url 填https://taotoken.net/api即可。
如果你只是想让 AI 帮你生成或补全 pymysql 序列化代码,可以直接用模型对话页面 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,把报错信息贴进去让它给方案。如果你是在做长期编码、Agent 类项目,建议看 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,额度模型更适合持续调用。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,配置细节以文档为准。
3. pymysql 连接配置与字段名还原
先解决最基础的一步:把查询结果从元组变成带字段名的字典。很多人卡在这里,是因为不知道cursor.description的结构。它是一个元组序列,每个元素的第一位[0]就是字段名。你可以用cursor.description配合zip或者索引遍历来还原。
下面是一段可复制的连接配置和字段名还原代码。连接参数用字典管理,方便你替换成自己的库信息:
import pymysql import json from datetime import datetime, date from decimal import Decimal DB_CONFIG = { "host": "127.0.0.1", "port": 3306, "user": "your_user", "password": "your_password", "database": "your_db", "charset": "utf8mb4", "cursorclass": pymysql.cursors.DictCursor, # 关键:直接返回字典 } def query_diaries(uid=3, limit=10): conn = pymysql.connect(**DB_CONFIG) try: with conn.cursor() as cursor: sql = "select id, title, content, created_at, price from diaries where uid=%s limit %s" cursor.execute(sql, (uid, limit)) rows = cursor.fetchall() return rows finally: conn.close()注意这里用了cursorclass=pymysql.cursors.DictCursor。这是最省事的方案,pymysql 会直接把每一行返回成{'id': 1, 'title': '...'}这样的字典,字段名自动带上,你不需要手动遍历cursor.description。如果你用的是默认的Cursor,那就得自己写还原逻辑:
def rows_to_dict(cursor, rows): columns = [desc[0] for desc in cursor.description] return [dict(zip(columns, row)) for row in rows]两种方式都行,DictCursor更简洁,手动还原更可控。我实测下来,DictCursor在字段多的时候性能略低一点点,但日常脚本完全无感,优先用它。
4. 自定义 JSONEncoder 处理 datetime 与 Decimal
字段名解决了,接下来是类型问题。datetime、date、Decimal这三类是报错重灾区。标准做法是继承json.JSONEncoder,重写default方法,遇到不认识的类型就转成字符串或数字。
class MySQLJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.strftime("%Y-%m-%d %H:%M:%S") if isinstance(obj, date): return obj.strftime("%Y-%m-%d") if isinstance(obj, Decimal): # 金额类建议转字符串避免浮点精度丢失 return str(obj) if isinstance(obj, bytes): return obj.decode("utf-8", errors="replace") if isinstance(obj, set): return list(obj) return super().default(obj)这里有个细节值得说:Decimal转float还是str?如果你做的是金额、价格类字段,强烈建议转str。因为float(Decimal('19.90'))可能变成19.9甚至19.899999999999999,前端展示和后续计算都会出问题。转字符串最安全,前端拿到后自己决定怎么处理。
datetime的格式也要统一。如果你的接口要给前端用,建议固定成"%Y-%m-%d %H:%M:%S",不要用isoformat()带T和时区偏移,除非你明确需要。统一格式能省掉前端一堆解析逻辑。
然后序列化的时候这样调用:
def to_json_string(rows): return json.dumps( rows, cls=MySQLJSONEncoder, ensure_ascii=False, # 中文不转义 indent=None, # 接口返回不要缩进,省带宽 separators=(",", ":"), )ensure_ascii=False是中文可读的关键。不加这个参数,《标题日记》会变成\u300a\u6807\u9898\u65e5\u8bb0\u300b。数据没错,但日志和调试体验差很多。加上之后,输出就是正常中文。
5. settings.json 中 TaoToken 统一 Key 配置片段
如果你用的是 Cline 这类 AI 编码工具,或者你的脚本需要调用模型接口做验证,settings.json里要配置 TaoToken 的统一 Key。下面是一个可复制的配置片段,字段名以你实际使用的工具为准,核心是base_url和api_key:
{ "taotoken": { "base_url": "https://taotoken.net/api", "api_key": "sk-your-taotoken-key-here", "model": "claude-sonnet-4-20250514", "timeout": 60, "max_retries": 2 }, "database": { "host": "127.0.0.1", "port": 3306, "user": "your_user", "password": "your_password", "database": "your_db", "charset": "utf8mb4" } }几个注意点。第一,base_url用https://taotoken.net/api,不要加 UTM 参数,UTM 是给网页追踪用的,API 端点加了反而可能出问题。第二,api_key不要硬编码在提交到 Git 的文件里,用环境变量或者.env读取。第三,model字段填你实际要用的模型名,具体可用模型以接入文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 为准。
读取配置的代码大概长这样:
import json import os def load_settings(path="settings.json"): with open(path, "r", encoding="utf-8") as f: cfg = json.load(f) # 优先用环境变量覆盖,避免 Key 泄露 cfg["taotoken"]["api_key"] = os.getenv("TAOTOKEN_API_KEY", cfg["taotoken"]["api_key"]) return cfg这样你在本地开发时把 Key 放环境变量,配置文件里留个占位符,提交代码就不会泄露。
6. 运行验证:从查询到 JSON 字符串的完整动作
把前面的代码串起来,写一个完整的验证脚本。假设你的diaries表有id、title、content、created_at、price五个字段,其中created_at是DATETIME,price是DECIMAL:
import pymysql import json from datetime import datetime, date from decimal import Decimal class MySQLJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.strftime("%Y-%m-%d %H:%M:%S") if isinstance(obj, date): return obj.strftime("%Y-%m-%d") if isinstance(obj, Decimal): return str(obj) return super().default(obj) def main(): conn = pymysql.connect( host="127.0.0.1", port=3306, user="your_user", password="your_password", database="your_db", charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) try: with conn.cursor() as cursor: cursor.execute( "select id, title, content, created_at, price from diaries where uid=%s limit %s", (3, 10), ) rows = cursor.fetchall() json_str = json.dumps(rows, cls=MySQLJSONEncoder, ensure_ascii=False) print(json_str) # 验证能否反序列化回来 parsed = json.loads(json_str) assert isinstance(parsed, list) assert parsed[0]["title"] is not None print("验证通过,共", len(parsed), "条记录") finally: conn.close() if __name__ == "__main__": main()运行后你应该看到类似这样的输出:
[{"id":1,"title":"《标题日记》完成输入功能","content":null,"created_at":"2025-01-15 09:30:00","price":"19.90"},{"id":2,"title":"睡了一天,有些累","content":null,"created_at":"2025-01-15 10:00:00","price":"0.00"}]中文正常显示,datetime变成了标准字符串,Decimal变成了字符串,字段名都在。最后用json.loads反序列化一遍,确认输出是合法 JSON,这一步很重要,能提前发现编码或类型问题。
如果你要把这个 JSON 字符串通过 TaoToken 的模型接口做进一步处理,比如让模型分析数据,可以在脚本里加一段请求:
import requests def ask_model(json_str, api_key): resp = requests.post( "https://taotoken.net/api/v1/chat/completions", headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, json={ "model": "claude-sonnet-4-20250514", "messages": [{"role": "user", "content": f"分析以下数据:{json_str}"}], }, timeout=60, ) return resp.json()请求地址用https://taotoken.net/api作为 base,具体路径以文档为准。验证模型对话效果可以直接在 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 里试。
7. 本篇常见错误排查
报错一:TypeError: Object of type datetime is not JSON serializable原因是没有用自定义 Encoder,或者 Encoder 里没处理datetime。检查json.dumps是否传了cls=MySQLJSONEncoder,以及default方法里isinstance判断是否覆盖了datetime。注意datetime是date的子类,判断顺序要先datetime后date,否则datetime会被date分支截胡,丢失时分秒。
报错二:TypeError: Object of type Decimal is not JSON serializable同上,Encoder 里加Decimal分支。如果你不想写 Encoder,也可以在 SQL 里用CAST(price AS CHAR)把DECIMAL转成字符串,但这样字段类型信息就丢了,不推荐。
报错三:JSON 里字段名丢失,只有数组你用的是默认Cursor,fetchall()返回元组,直接json.dumps就变成数组了。解决方式二选一:用DictCursor,或者手动用cursor.description还原字段名。手动还原的代码在第三节给了。
报错四:中文变成\uXXXX转义json.dumps默认ensure_ascii=True。加上ensure_ascii=False即可。注意这个参数只影响输出,不影响数据本身。
报错五:json.loads报Expecting value或解析失败检查你的 JSON 字符串里有没有NaN、Infinity这类非标准值。MySQL 的FLOAT字段如果存了异常值,json.dumps默认会输出NaN,这不是合法 JSON。可以在 Encoder 里把float('nan')转成None,或者用json.dumps(..., allow_nan=False)让它直接报错,提前暴露问题。
报错六:连接超时或Access denied检查DB_CONFIG里的 host、port、user、password、database 是否和实际一致。charset建议固定utf8mb4,避免中文和 emoji 乱码。如果 Key 相关配置有问题,去 API Keys 页面 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 确认 Key 状态,接入细节看文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。
8. 继续把这条链路用起来
到这里,pymysql 查询结果转 JSON 字符串的核心链路已经跑通了:连接配置、字段名还原、自定义 Encoder、中文处理、运行验证、错误排查,每一步都有可复制的代码。你可以直接把第六节的完整脚本拿去改表名和字段名,五分钟就能跑起来。
如果你后续要做的是长期编码项目,或者让 AI Agent 持续帮你生成和调试这类数据接口,建议把 TaoToken 的 Coding Plan 配起来 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,统一 Key 通道省掉多平台切换的麻烦。日常快速验证模型输出,用模型对话页面 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 就够了。控制台和 Key 管理分别在 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 和 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。
最后留一个实用技巧:把MySQLJSONEncoder单独放到一个utils/json_encoder.py文件里,所有脚本共用。下次遇到datetime报错,直接from utils.json_encoder import MySQLJSONEncoder,不用每次重写。这个类我用了两年多,覆盖了 MySQL 常见类型的 95% 场景,剩下 5% 是GEOMETRY和JSON字段,遇到再单独处理就行。