1. 从一次金额对不上账说起pymysql 类型转换到底坑在哪如果你用 Python 通过 pymysql 读写 MySQL大概率遇到过这种诡异现象数据库里明明存的是99.99Python 里打印出来却是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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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( hostconfig[host], portconfig[port], userconfig[user], passwordconfig[password], databaseconfig[database], charsetconfig[charset], cursorclasspymysql.cursors.DictCursor, convconv, autocommitFalse, ) return conn这里有几个参数值得单独说。cursorclasspymysql.cursors.DictCursor让查询结果以字典返回字段名做键比默认的元组好读太多后面做类型校验也方便。autocommitFalse是显式关闭自动提交写操作后必须手动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 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 ) 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_asciiFalse), 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而不是floatprofile是str需要你自己json.loadsraw_data是bytes。如果amount打印出来是float说明conv没生效检查是不是在connect()之后才改的映射表——映射必须在连接建立前传入。如果你想让模型帮你分析这段脚本的输出或报错可以把结果贴到 TaoToken 的模型对话里走统一 Key 通道https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。长期做编码和 Agent 任务的话Coding Plan 更适合入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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_asciiFalse否则中文会变成\uXXXX转义虽然能存能读但可读性差排查问题时很痛苦。5.4 参数化查询与类型拼接永远用%s占位符不要用 f-string 或拼接 SQL。pymysql 的占位符不只是防注入它还负责把 Python 类型正确转成 MySQL 字面量。你手动str(Decimal(99.99))拼进去看似没问题但遇到None、datetime、bytes时就会拼出非法 SQL。这个原则在类型转换场景里尤其重要因为转换逻辑本来就复杂再叠加字符串拼接只会让问题更难定位。5.5 连接字符集与排序规则charsetutf8mb4要写在连接参数里不要只在建表时写。连接层的字符集决定了传输编码建表层决定存储编码两者不一致时中文可能变成???。如果已经建了表且排序规则是utf8mb4_general_ci连接层用utf8mb4即可不需要额外指定collation。6. 把类型转换收进一层别散落在业务代码里类型转换这件事最怕的不是不会转而是转的地方太多。今天在查询后float()一下明天在写入前str()一下三个月后没人记得哪个字段被转过。我的做法是所有数据库读写都经过一个薄薄的 repository 层转换规则集中在那里业务代码只拿「已经转好的 Python 原生类型」。具体来说DECIMAL统一在连接层转DecimalJSON统一在 repository 层json.loadsdatetime保持datetime对象不提前格式化只在最终输出给前端或写文件时才strftime。这样每一层的职责是清晰的连接层管驱动级映射repository 层管业务语义转换业务层只管用。如果你在接入或排障过程中需要查具体的 API 参数和字段说明接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。把 Key 和数据库密码都放进settings.json并加进.gitignore这是最省心的收尾方式。 Python 操作 MySQL 数据类型转换实战：用 pymysql 处理 Decimal 与 TaoToken 配置
1. 从一次金额对不上账说起pymysql 类型转换到底坑在哪如果你用 Python 通过 pymysql 读写 MySQL大概率遇到过这种诡异现象数据库里明明存的是99.99Python 里打印出来却是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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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( hostconfig[host], portconfig[port], userconfig[user], passwordconfig[password], databaseconfig[database], charsetconfig[charset], cursorclasspymysql.cursors.DictCursor, convconv, autocommitFalse, ) return conn这里有几个参数值得单独说。cursorclasspymysql.cursors.DictCursor让查询结果以字典返回字段名做键比默认的元组好读太多后面做类型校验也方便。autocommitFalse是显式关闭自动提交写操作后必须手动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 ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 ) 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_asciiFalse), 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而不是floatprofile是str需要你自己json.loadsraw_data是bytes。如果amount打印出来是float说明conv没生效检查是不是在connect()之后才改的映射表——映射必须在连接建立前传入。如果你想让模型帮你分析这段脚本的输出或报错可以把结果贴到 TaoToken 的模型对话里走统一 Key 通道https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。长期做编码和 Agent 任务的话Coding Plan 更适合入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_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_asciiFalse否则中文会变成\uXXXX转义虽然能存能读但可读性差排查问题时很痛苦。5.4 参数化查询与类型拼接永远用%s占位符不要用 f-string 或拼接 SQL。pymysql 的占位符不只是防注入它还负责把 Python 类型正确转成 MySQL 字面量。你手动str(Decimal(99.99))拼进去看似没问题但遇到None、datetime、bytes时就会拼出非法 SQL。这个原则在类型转换场景里尤其重要因为转换逻辑本来就复杂再叠加字符串拼接只会让问题更难定位。5.5 连接字符集与排序规则charsetutf8mb4要写在连接参数里不要只在建表时写。连接层的字符集决定了传输编码建表层决定存储编码两者不一致时中文可能变成???。如果已经建了表且排序规则是utf8mb4_general_ci连接层用utf8mb4即可不需要额外指定collation。6. 把类型转换收进一层别散落在业务代码里类型转换这件事最怕的不是不会转而是转的地方太多。今天在查询后float()一下明天在写入前str()一下三个月后没人记得哪个字段被转过。我的做法是所有数据库读写都经过一个薄薄的 repository 层转换规则集中在那里业务代码只拿「已经转好的 Python 原生类型」。具体来说DECIMAL统一在连接层转DecimalJSON统一在 repository 层json.loadsdatetime保持datetime对象不提前格式化只在最终输出给前端或写文件时才strftime。这样每一层的职责是清晰的连接层管驱动级映射repository 层管业务语义转换业务层只管用。如果你在接入或排障过程中需要查具体的 API 参数和字段说明接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content Key 管理在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。把 Key 和数据库密码都放进settings.json并加进.gitignore这是最省心的收尾方式。