1. 从一次金额对不上账说起:pymysql 类型转换到底坑在哪
如果你用 Python 通过 pymysql 读写 MySQL,大概率遇到过这种诡异现象:数据库里明明存的是99.99,Python 里打印出来却是99.98999999999999;或者接口返回的 JSON 里,create_time是个datetime.datetime对象,json.dumps直接抛TypeError: Object of type datetime is not JSON serializable。这些都不是数据库坏了,而是 Python 与 MySQL 之间的数据类型映射在作怪。
pymysql 作为纯 Python 实现的 MySQL 客户端,默认会帮你做一层类型转换:INT变int、VARCHAR变str、DATETIME变datetime.datetime、NULL变None。这套默认规则能覆盖大约八成的日常场景,但一旦碰到金额、汇率、时间戳格式、JSON 字段、二进制文件,默认行为就会给你埋雷。尤其是DECIMAL类型,pymysql 默认把它转成float,而浮点数在二进制里无法精确表示0.1这类十进制小数,做累加或对账时误差会被放大。
这篇内容聚焦的就是这些「类型不一致」的典型报错场景,我会把可复制的 pymysql 连接配置、TaoToken 统一 Key/API 通道的settings.json骨架,以及查询结果类型校验与转换的验证脚本都摊开讲。适合已经会写基础 CRUD、但被类型转换卡住的 Python 后端或数据开发者。读完之后,你应该能自己定位「这个字段为什么变成了 float」「这个 datetime 为什么序列化失败」,并且知道该在哪一层拦截转换。
2. 前置准备:TaoToken 统一 Key 与 pymysql 环境
在动手改转换规则之前,先把「模型调用通道」和「数据库连接」这两件事分开理清。很多同学在调试类型转换时,顺手把 AI 辅助生成的代码片段直接贴进项目,结果 Key 散落在各个脚本里,后面排查问题时根本分不清哪段代码用了哪个通道。我的做法是:数据库连接走本地配置,模型调用走 TaoToken 的统一 Key,两者互不污染。
TaoToken 在这里的角色是提供一个统一的 API 通道,让你在写转换脚本、让模型帮忙分析报错时,不用在多个平台之间来回切换 Key。官网入口是 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,API 基址是 https://taotoken.net/api ,注意 API 地址后面不加 UTM 参数,保持干净。
环境侧你需要两样东西:一是本地或远程可连的 MySQL 5.7 及以上版本(JSON 类型需要 5.7+),二是 Python 3.8+ 和 pymysql。安装命令很直接:
pip install pymysql cryptographycryptography是 pymysql 连接 MySQL 8 默认认证插件caching_sha2_password时的依赖,不装会在握手阶段报RuntimeError: 'cryptography' package is required。这个报错我见过太多次,很多人以为是密码错了,其实是缺包。
TaoToken 的 Key 建议放在项目根目录的settings.json里,和数据库配置并列但分节管理。骨架长这样:
{ "taotoken": { "api_base": "https://taotoken.net/api", "api_key": "sk-你的统一Key", "default_model": "claude-sonnet-4-20250514" }, "mysql": { "host": "127.0.0.1", "port": 3306, "user": "app_user", "password": "你的数据库密码", "database": "demo_db", "charset": "utf8mb4" } }charset一定要写utf8mb4,不要写utf8。MySQL 的utf8是三字节的残废实现,存 emoji 或部分生僻字会直接报Incorrect string value。这个坑和类型转换无关,但经常和类型报错混在一起出现,先排掉。
3. 可复制配置:重写 pymysql 的 converters 映射表
pymysql 的类型转换核心在pymysql.converters.conversions这个字典里,键是 MySQL 的字段类型常量,值是解码函数。默认情况下FIELD_TYPE.DECIMAL和FIELD_TYPE.NEWDECIMAL都指向float。我们要做的就是在建立连接前复制这份映射,把这两个键改成Decimal。
import pymysql from decimal import Decimal from pymysql.constants import FIELD_TYPE def build_connection(config: dict): conv = pymysql.converters.conversions.copy() # DECIMAL 精准转换,避免 float 精度丢失 conv[FIELD_TYPE.DECIMAL] = Decimal conv[FIELD_TYPE.NEWDECIMAL] = Decimal # 可选:把 TINYINT(1) 当布尔处理,视业务而定 # conv[FIELD_TYPE.TINY] = bool conn = pymysql.connect( host=config["host"], port=config["port"], user=config["user"], password=config["password"], database=config["database"], charset=config["charset"], cursorclass=pymysql.cursors.DictCursor, conv=conv, autocommit=False, ) return conn这里有几个参数值得单独说。cursorclass=pymysql.cursors.DictCursor让查询结果以字典返回,字段名做键,比默认的元组好读太多,后面做类型校验也方便。autocommit=False是显式关闭自动提交,写操作后必须手动commit(),避免调试脚本时误写数据。
关于TINYINT(1)要不要转bool,我的建议是看业务语义。MySQL 没有真正的布尔类型,BOOLEAN只是TINYINT(1)的别名。如果你的字段叫is_deleted、is_active,转bool更符合直觉;但如果字段叫status且取值是 0/1/2,转bool会把 2 也变成True,反而制造 bug。所以上面代码里我把它注释掉了,按需开启。
日期时间这块,pymysql 默认把DATETIME、TIMESTAMP、DATE分别转成datetime.datetime和datetime.date,这个默认行为通常是对的,不建议改。真正需要处理的是「读出来之后怎么序列化」和「写进去之前怎么格式化」,这属于应用层的事,放在第 4 节的验证脚本里做。
JSON 类型要特别注意:pymysql 不会自动把 MySQL 的 JSON 字段转成 Python 字典,它返回的是str。所以写入时你要json.dumps,读取时你要json.loads。如果你用的是 MySQL 8 且驱动版本较新,某些情况下会返回bytes,这时要先decode('utf-8')再json.loads。稳妥的写法是统一判断类型:
import json def parse_json_field(value): if value is None: return None if isinstance(value, bytes): value = value.decode("utf-8") if isinstance(value, str): return json.loads(value) return value4. 验证请求:类型校验与转换脚本
配置写好了,得有个脚本能跑起来验证每个字段的真实类型。下面这个脚本会建一张覆盖DECIMAL、DATETIME、JSON、BIGINT、BLOB的测试表,插入数据后逐字段打印类型,并做一次「读出来再写回去」的往返测试。
import json from datetime import datetime from decimal import Decimal from build_conn import build_connection # 上一节的函数 config = { "host": "127.0.0.1", "port": 3306, "user": "app_user", "password": "你的数据库密码", "database": "demo_db", "charset": "utf8mb4", } conn = build_connection(config) cursor = conn.cursor() cursor.execute(""" CREATE TABLE IF NOT EXISTS type_demo ( id BIGINT AUTO_INCREMENT PRIMARY KEY, order_no VARCHAR(32), amount DECIMAL(12, 2), created_at DATETIME, profile JSON, raw_data BLOB ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) profile = {"hobby": ["篮球", "编程"], "city": "北京"} cursor.execute( "INSERT INTO type_demo (order_no, amount, created_at, profile, raw_data) " "VALUES (%s, %s, %s, %s, %s)", ("ORD202501", Decimal("99.99"), datetime(2025, 1, 19, 10, 30, 0), json.dumps(profile, ensure_ascii=False), b"\x00\x01\x02"), ) conn.commit() cursor.execute("SELECT * FROM type_demo WHERE order_no = %s", ("ORD202501",)) row = cursor.fetchone() checks = { "amount": (row["amount"], Decimal), "created_at": (row["created_at"], datetime), "profile": (row["profile"], str), "raw_data": (row["raw_data"], bytes), } for field, (value, expect) in checks.items(): ok = isinstance(value, expect) print(f"{field}: value={value!r} type={type(value).__name__} expect={expect.__name__} pass={ok}") # 往返:读出来再写回去,验证转换不丢精度 cursor.execute( "UPDATE type_demo SET amount = %s WHERE order_no = %s", (row["amount"], "ORD202501"), ) conn.commit() cursor.execute("SELECT amount FROM type_demo WHERE order_no = %s", ("ORD202501",)) print("往返后金额:", cursor.fetchone()["amount"]) cursor.close() conn.close()跑通后你会看到amount的类型是Decimal而不是float,profile是str(需要你自己json.loads),raw_data是bytes。如果amount打印出来是float,说明conv没生效,检查是不是在connect()之后才改的映射表——映射必须在连接建立前传入。
如果你想让模型帮你分析这段脚本的输出或报错,可以把结果贴到 TaoToken 的模型对话里,走统一 Key 通道:https://taotoken.net/api?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= 。
5. 本篇常见错排查
5.1 Decimal 还是 float:精度丢失的根因
最常见的报错不是抛异常,而是「静默错误」——金额累加后差几分钱。根因就是DECIMAL被转成了float。判断方法很简单:print(type(row['amount'])),如果是float,立刻去检查conv映射。另一个容易忽略的点是:即使读取时转成了Decimal,如果你在 Python 里用float(row['amount'])再参与计算,精度照样丢。全程保持Decimal运算,最后输出时再格式化。
5.2 datetime 序列化失败
TypeError: Object of type datetime is not JSON serializable这个报错,出现在你把查询结果直接json.dumps的时候。解决办法有两个:一是自定义JSONEncoder,二是查询后手动strftime。我倾向后者,因为格式可控:
class DateTimeEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, datetime): return obj.strftime("%Y-%m-%d %H:%M:%S") if isinstance(obj, Decimal): return str(obj) return super().default(obj)注意Decimal也要处理,否则json.dumps同样会抛TypeError。把Decimal转成str而不是float,是为了保住精度。
5.3 JSON 字段返回 bytes 或 str 不确定
不同 MySQL 版本和 pymysql 版本下,JSON 字段的返回类型可能是str也可能是bytes。不要假设,用第 3 节的parse_json_field统一处理。另外写入时json.dumps记得加ensure_ascii=False,否则中文会变成\uXXXX转义,虽然能存能读,但可读性差,排查问题时很痛苦。
5.4 参数化查询与类型拼接
永远用%s占位符,不要用 f-string 或+拼接 SQL。pymysql 的占位符不只是防注入,它还负责把 Python 类型正确转成 MySQL 字面量。你手动str(Decimal("99.99"))拼进去,看似没问题,但遇到None、datetime、bytes时就会拼出非法 SQL。这个原则在类型转换场景里尤其重要,因为转换逻辑本来就复杂,再叠加字符串拼接只会让问题更难定位。
5.5 连接字符集与排序规则
charset='utf8mb4'要写在连接参数里,不要只在建表时写。连接层的字符集决定了传输编码,建表层决定存储编码,两者不一致时中文可能变成???。如果已经建了表且排序规则是utf8mb4_general_ci,连接层用utf8mb4即可,不需要额外指定collation。
6. 把类型转换收进一层,别散落在业务代码里
类型转换这件事,最怕的不是不会转,而是转的地方太多。今天在查询后float()一下,明天在写入前str()一下,三个月后没人记得哪个字段被转过。我的做法是:所有数据库读写都经过一个薄薄的 repository 层,转换规则集中在那里,业务代码只拿「已经转好的 Python 原生类型」。
具体来说,DECIMAL统一在连接层转Decimal,JSON统一在 repository 层json.loads,datetime保持datetime对象不提前格式化,只在最终输出给前端或写文件时才strftime。这样每一层的职责是清晰的:连接层管驱动级映射,repository 层管业务语义转换,业务层只管用。
如果你在接入或排障过程中需要查具体的 API 参数和字段说明,接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,Key 管理在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。把 Key 和数据库密码都放进settings.json并加进.gitignore,这是最省心的收尾方式。