1. 从一次数据导出需求说起
Python3 提取 MySQL 数据并转成字典数组,是后端脚本、数据清洗、接口返回组装里出现频率极高的一步。你查完库拿到的是元组列表,但下游要的是[{"id":1,"name":"..."}]这种结构,中间这层转换如果写得不稳,字段顺序、列名大小写、空值处理都会埋坑。这篇就聚焦这条完整链路:连接参数怎么配、cursor 类型怎么选、字段映射怎么写、返回结构怎么验证,最后给一份可复制的config.toml骨架和统一 Key 配置示例。
适合谁看:刚接触 pymysql 想一次写对的同学;已经能查库但返回结构总对不上的同学;想把数据库连接信息从代码里抽出来、用统一配置管理的同学。核心检索词就三个:python3、mysql、字典数组。下面所有代码我都实际跑过,参数和报错都来自真实终端输出。
先说结论:转字典数组最稳的写法是cursor.description取列名,再配合zip和字典推导式,一行搞定,不用手写索引循环。但前提是连接和 cursor 配置要对,否则description拿不到、中文乱码、字段名重复这些问题会接连出现。
2. TaoToken 统一 Key 前置配置
在写数据库脚本之前,先把模型调用这条线的前置配置理清。很多同学的数据脚本后面要接一个「把查询结果丢给模型做摘要/分类」的步骤,这时候如果每个脚本都硬编码一份 Key,维护起来很痛苦。TaoToken 提供统一 Key 的方式,把模型访问收敛到一个入口,脚本里只读配置。
官网入口:https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=
API 地址:https://taotoken.net/api
你需要先拿到 Key,再去控制台确认额度与模型可用性。这几个 deep link 按用途分:
- 模型对话(验证模型是否通):https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=model-chat
- Coding Plan(长期编码/Agent 场景):https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=coding-plan
- 控制台:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=console
- API Keys 管理:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=api-keys
- 接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc
- ClaudeCodeAnthropic 相关:https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=claudecode
注意:Key 只放在本地配置文件或环境变量里,不要提交到 Git。下面
config.toml里的占位符记得替换成你自己的值。
3. 可复制的 config.toml 与连接配置
3.1 config.toml 骨架
把数据库连接和模型 Key 都收进一个 TOML 文件,Python3 用标准库tomllib(3.11+)或tomli读取。骨架如下:
# config.toml [mysql] host = "127.0.0.1" port = 3306 user = "root" password = "your_password" database = "data" charset = "utf8mb4" [taotoken] api_base = "https://taotoken.net/api" api_key = "sk-替换成你的Key" model = "gpt-4o-mini"charset建议用utf8mb4,比utf8多支持 emoji 和部分生僻字,避免写入时报Incorrect string value。
3.2 读取配置并建立连接
import tomllib # Python 3.11+;低版本用 pip install tomli 后 import tomli as tomllib import pymysql with open("config.toml", "rb") as f: cfg = tomllib.load(f) mysql_cfg = cfg["mysql"] conn = pymysql.connect( host=mysql_cfg["host"], port=mysql_cfg["port"], user=mysql_cfg["user"], password=mysql_cfg["password"], database=mysql_cfg["database"], charset=mysql_cfg["charset"], cursorclass=pymysql.cursors.DictCursor, # 关键:直接返回字典 )这里有个关键选择:cursorclass。pymysql 默认是Cursor,fetchall()返回元组列表;设成DictCursor后,每行直接是字典,列名做 key。两种路线:
| cursor 类型 | fetchall 返回 | 转字典数组成本 | 适用场景 |
|---|---|---|---|
Cursor(默认) | 元组列表 | 需手动 zip 列名 | 需要按位置取值、列名重复 |
DictCursor | 字典列表 | 几乎为零 | 绝大多数业务查询 |
如果你用DictCursor,其实第 4 节的转换代码可以省掉,但很多老项目、或者需要兼容多种 cursor 的场景,还是得会手动转。下面两种都讲。
4. 查询结果转字典数组的三种写法
4.1 手动 zip 列名(推荐)
这是最通用、最好理解的写法,不依赖DictCursor:
import pymysql conn = pymysql.connect( host="127.0.0.1", port=3306, user="root", password="your_password", database="data", charset="utf8mb4", ) sql = "SELECT id, name, certificate_no FROM userinfo" cursor = conn.cursor() try: cursor.execute(sql) columns = [col[0] for col in cursor.description] # 取列名 rows = cursor.fetchall() data_dict = [dict(zip(columns, row)) for row in rows] print(data_dict) except Exception as e: conn.rollback() print("查询失败:", e) finally: cursor.close() conn.close()cursor.description是一个元组序列,每个元素形如('id', None, None, None, None, None, None),第 0 位就是列名。zip(columns, row)把列名和值配对,dict()转成字典,外层列表推导式把每行都转一遍,最终就是字典数组。
4.2 用 DictCursor 直接拿
如果你在连接时已经设了cursorclass=pymysql.cursors.DictCursor,那fetchall()本身就是字典数组:
cursor = conn.cursor() cursor.execute("SELECT id, name FROM userinfo") data_dict = cursor.fetchall() # 直接就是 [{'id':1,'name':'...'}]省事,但要注意:DictCursor下如果 SQL 里有同名列(比如 join 两张表都有id),后面的会覆盖前面的,得用别名区分。
4.3 迭代函数 + map(了解即可)
有些老代码用map加自定义函数:
def row_to_dict(row): return dict(zip([c[0] for c in cursor.description], row)) data_dict = list(map(row_to_dict, cursor.fetchall()))功能等价,但可读性不如列表推导式,而且cursor.description在函数里是闭包引用,容易踩作用域坑。新代码不建议这么写。
5. 验证返回结构是否为字典数组
写完转换逻辑,别急着往下游传,先验证结构。我习惯加一段断言式检查:
def validate_dict_array(data): assert isinstance(data, list), "顶层必须是 list" for i, row in enumerate(data): assert isinstance(row, dict), f"第 {i} 行不是 dict" for k in row: assert isinstance(k, str), f"第 {i} 行 key 不是字符串: {k}" return True result = [dict(zip(columns, row)) for row in rows] print("结构校验:", validate_dict_array(result)) print("首行样例:", result[0] if result else "空结果")跑一次真实输出大概是这样:
结构校验: True 首行样例: {'id': 1, 'name': '张三', 'certificate_no': '20200317001'}如果查询结果为空,result是[],校验也通过,但result[0]会 IndexError,所以打印样例前要判空。这一步能提前拦住「返回了元组列表却当成字典用」的低级错误。
6. 本篇常见错排查
6.1 报错'tuple' object has no attribute 'get'
说明你拿到的是元组列表,却按字典用了。检查连接是否漏了cursorclass=DictCursor,或者转换那步没执行。用第 5 节的validate_dict_array一测就现形。
6.2 中文乱码
连接charset写utf8有时不够,改成utf8mb4。另外确认表本身的字符集也是utf8mb4,否则连接对了、表不对,照样乱码。
6.3cursor.description为 None
只有执行过execute之后description才有值。如果你在execute前就取列名,拿到的是None,后面zip会报错。顺序必须是:execute→ 取description→fetchall。
6.4 字段名重复导致覆盖
join 查询里两张表都有id,DictCursor或zip转字典时后者覆盖前者。解决办法是 SQL 里用别名:SELECT a.id AS a_id, b.id AS b_id。
6.5 忘记关闭连接
用try/finally或with上下文管理,确保cursor.close()和conn.close()执行。连接泄漏在长时间运行的脚本里会耗尽连接数。
7. 下一步:把结果接给模型
字典数组拿到后,如果你要把它丢给模型做摘要、打标签、生成报告,就用第 2 节配好的统一 Key。脚本里读config.toml的[taotoken]段,请求发到https://taotoken.net/api,模型名按需替换。验证模型是否通,直接去模型对话页发一条测试消息最快:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=model-chat
长期跑编码或 Agent 任务,用 Coding Plan 更省心:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=coding-plan
Key 管理和额度查看在控制台:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=console
接入细节和参数说明看文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc
最后留一个我踩过的坑:config.toml里的密码如果含#或",TOML 解析会出问题,用单引号包裹或转义。数据库连接和模型 Key 都收进配置文件后,脚本本身就能干净地只做「查询 → 转字典数组 → 校验 → 下游处理」这一条链路,换环境只改配置不改代码。