pymysql 获取查询字段名（列名）：TaoToken 统一 Key 下的配置骨架与验证动作

1. pymysql 查询拿不到列名到底卡在哪用 pymysql 做数据查询时cursor.fetchall()返回的只是行数据列名信息并不在结果集里。这个问题在写通用导出工具、动态报表、数据同步脚本时特别明显你不知道每一列叫什么就没法把结果映射成字典也没法做字段级别的校验和转换。cursor.description是 pymysql 提供的元数据入口。执行cursor.execute(sql)之后cursor.description会变成一个元组序列每个元素描述一列其中第 0 个位置就是列名或别名。很多人第一次看到((ERROR_CODE, 253, None, 92, 92, 0, True), ...)这种结构会懵其实它对应的是 DB-API 规范里的 7 元组name、type_code、display_size、internal_size、precision、scale、null_ok。我们只需要col[i][0]。这篇内容聚焦工程落地在 TaoToken 统一 Key/API 通道下给出config.toml与settings.json的可复制骨架演示用cursor.description提取字段名、映射为字典列表并附一条可执行的验证脚本确认列名顺序与别名解析正确。适合正在写数据管道、ETL 小工具、或者需要把查询结果转成 JSON 返回给前端的同学。核心检索词先摆出来pymysql 获取查询字段名、cursor.description 列名、pymysql 列名转字典、pymysql 别名解析。这几个词覆盖了从“怎么拿”到“怎么用”的完整链路。2. TaoToken 前置统一 Key 与配置骨架TaoToken 在这里的角色是统一 API 通道。你可能会问pymysql 是直连数据库的跟 API 通道有什么关系实际工程里数据库查询往往只是链路的一环查询结果要送到模型做语义解析、字段映射、异常归因这时候就需要一个稳定的 API 出口。TaoToken 提供统一 Key把模型对话、coding-plan、console、api-keys 等入口收敛到一套凭证体系里省得每个服务各配一套。官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基址https://taotoken.net/api需要先拿到 Key 的话走这个 deep linkhttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite模型对话入口验证字段语义映射时用https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite长期做编码和 Agent 的话Coding Plan 更合适https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewriteConsole 控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewriteClaudeCodeAnthropic 入口https://taotoken.net/claude-code-anthropic?utm_sourcetaotoken_aicg_blog_endutm_contentclaude_codeutm_campaignrewrite注意TaoToken 是统一 API 通道不是数据库代理。pymysql 仍然直连你的 MySQLTaoToken 负责的是查询结果后续的模型调用与凭证管理。3. 可复制配置config.toml 与 settings.json先把配置骨架搭好。我习惯把数据库连接和 API 通道分开管理数据库走config.tomlAPI 凭证走settings.json这样换环境时只改一个文件。3.1 config.toml 数据库连接骨架# config.toml [mysql] host 127.0.0.1 port 3308 user your_user password your_password database your_db charset utf8mb4 connect_timeout 10 read_timeout 30 [query] default_limit 1000 fetch_mode dict # dict | tuple alias_strict true # 是否严格校验别名 [taotoken] api_base https://taotoken.net/api model gpt-4o-mini timeout 60fetch_mode控制返回格式alias_strict控制别名解析时是否报错。这两个参数后面验证脚本会用到。3.2 settings.json API 凭证骨架{ taotoken: { api_key: sk-xxxxxxxxxxxxxxxx, api_base: https://taotoken.net/api, default_model: gpt-4o-mini, max_retries: 3, retry_backoff: 1.5 }, logging: { level: INFO, file: logs/pymysql_columns.log } }提示api_key不要硬编码进代码仓库用环境变量注入更稳妥。settings.json 只放结构实际值从TAOTOKEN_API_KEY读取。3.3 配置加载器# config_loader.py import json import os import tomllib from pathlib import Path def load_toml(path: str config.toml) - dict: with open(path, rb) as f: return tomllib.load(f) def load_settings(path: str settings.json) - dict: with open(path, r, encodingutf-8) as f: cfg json.load(f) # 环境变量覆盖 env_key os.getenv(TAOTOKEN_API_KEY) if env_key: cfg[taotoken][api_key] env_key return cfg if __name__ __main__: db_cfg load_toml()[mysql] api_cfg load_settings()[taotoken] print(mysql host:, db_cfg[host]) print(api base:, api_cfg[api_base])跑一下确认配置能读出来python config_loader.py # 输出 # mysql host: 127.0.0.1 # api base: https://taotoken.net/api4. cursor.description 提取列名与字典映射配置就绪后进入核心部分。cursor.description的结构是 7 元组序列我们只取第 0 位。下面给出完整实现包含列名提取、字典列表映射、别名解析。4.1 基础列名提取# column_utils.py import pymysql def get_columns(cursor) - list[str]: 从 cursor.description 提取列名列表保持顺序 if cursor.description is None: return [] return [col[0] for col in cursor.description] def query_with_columns(conn, sql: str): 执行查询返回 (列名列表, 行数据) with conn.cursor() as cursor: cursor.execute(sql) columns get_columns(cursor) rows cursor.fetchall() return columns, rowscursor.description在execute之后才有值fetchall之前取最稳妥。如果 SQL 是 DDL 或没有结果集description会是None所以加了判空。4.2 映射为字典列表# dict_mapper.py def rows_to_dicts(columns: list[str], rows: tuple) - list[dict]: 把行数据映射为字典列表列名做 key return [dict(zip(columns, row)) for row in rows] def rows_to_dicts_safe(columns: list[str], rows: tuple) - list[dict]: 带长度校验的版本防止列数与行长度不一致 result [] for row in rows: if len(row) ! len(columns): raise ValueError( f列数({len(columns)})与行长度({len(row)})不匹配 ) result.append(dict(zip(columns, row))) return resultzip在列名和行长度不一致时会静默截断生产环境建议用rows_to_dicts_safe。4.3 别名解析验证SQL 里用AS起别名时cursor.description返回的是别名而不是原字段名。这一点很关键很多人以为拿到的是原始列名。# alias_check.py import pymysql SQL_WITH_ALIAS SELECT event_id AS 事件ID, comment_sql AS 预警说明, metric_rate AS 指标率 FROM etl_event_head WHERE event_id 6001 def check_alias(conn): with conn.cursor() as cursor: cursor.execute(SQL_WITH_ALIAS) cols [c[0] for c in cursor.description] rows cursor.fetchall() print(列名顺序:, cols) print(首行:, rows[0] if rows else None) return cols, rows执行后输出类似列名顺序: [事件ID, 预警说明, 指标率] 首行: (6001, 联系人→进件, 6%)可以看到别名完全覆盖了原始字段名顺序与 SELECT 子句一致。这就是alias_strict参数要校验的东西如果你期望拿到event_id但实际拿到事件ID说明别名解析生效了。4.4 完整查询封装# query_engine.py import pymysql from config_loader import load_toml, load_settings from column_utils import get_columns from dict_mapper import rows_to_dicts_safe class QueryEngine: def __init__(self): self.db_cfg load_toml()[mysql] self.query_cfg load_toml()[query] self.api_cfg load_settings()[taotoken] self.conn None def connect(self): self.conn pymysql.connect( hostself.db_cfg[host], portself.db_cfg[port], userself.db_cfg[user], passwordself.db_cfg[password], databaseself.db_cfg[database], charsetself.db_cfg[charset], connect_timeoutself.db_cfg[connect_timeout], read_timeoutself.db_cfg[read_timeout], cursorclasspymysql.cursors.Cursor, ) return self.conn def query(self, sql: str) - dict: if self.conn is None: self.connect() with self.conn.cursor() as cursor: cursor.execute(sql) columns get_columns(cursor) rows cursor.fetchall() if self.query_cfg[fetch_mode] dict: data rows_to_dicts_safe(columns, rows) else: data list(rows) return {columns: columns, rows: data, count: len(data)} def close(self): if self.conn: self.conn.close() self.conn None调用方式engine QueryEngine() result engine.query(SELECT event_id, comment_sql FROM etl_event_head LIMIT 5) print(result[columns]) print(result[rows][0]) engine.close()5. 验证请求与成功结果配置和代码都齐了跑一条验证脚本确认列名顺序与别名解析正确。这个脚本会做三件事连接数据库、执行带别名的查询、校验列名与预期一致。5.1 验证脚本# verify_columns.py import sys from query_engine import QueryEngine EXPECTED_COLUMNS [事件ID, 预警说明, 指标率] SQL SELECT event_id AS 事件ID, comment_sql AS 预警说明, metric_rate AS 指标率 FROM etl_event_head WHERE event_id 6001 def main(): engine QueryEngine() try: result engine.query(SQL) cols result[columns] print(实际列名:, cols) print(预期列名:, EXPECTED_COLUMNS) if cols ! EXPECTED_COLUMNS: print(FAIL: 列名或顺序不匹配) sys.exit(1) print(PASS: 列名顺序与别名解析正确) print(首行数据:, result[rows][0] if result[rows] else 无数据) finally: engine.close() if __name__ __main__: main()5.2 执行结果python verify_columns.py # 输出 # 实际列名: [事件ID, 预警说明, 指标率] # 预期列名: [事件ID, 预警说明, 指标率] # PASS: 列名顺序与别名解析正确 # 首行数据: {事件ID: 6001, 预警说明: 联系人→进件, 指标率: 6%}5.3 接入 TaoToken 做字段语义校验拿到列名后可以把列名列表送到 TaoToken 的模型对话接口做字段语义映射校验。比如确认“指标率”是否对应业务口径里的metric_rate。# semantic_check.py import json import urllib.request from config_loader import load_settings def check_semantics(columns: list[str]) - str: cfg load_settings()[taotoken] prompt ( 以下是一组数据库查询列名请判断它们是否语义清晰、 是否存在歧义或命名冲突用一句话给出结论\n json.dumps(columns, ensure_asciiFalse) ) payload json.dumps({ model: cfg[default_model], messages: [{role: user, content: prompt}], temperature: 0.2, }).encode(utf-8) req urllib.request.Request( f{cfg[api_base]}/v1/chat/completions, datapayload, headers{ Content-Type: application/json, Authorization: fBearer {cfg[api_key]}, }, methodPOST, ) with urllib.request.urlopen(req, timeoutcfg[timeout]) as resp: body json.loads(resp.read().decode(utf-8)) return body[choices][0][message][content] if __name__ __main__: cols [事件ID, 预警说明, 指标率] print(check_semantics(cols))这一步不是必须的但在字段命名混乱的老系统里用模型做一轮语义体检能省不少人工核对时间。6. 本篇常见错排查6.1 cursor.description 返回 None最常见的原因是 SQL 不是查询语句或者execute还没执行。DDL、INSERT、UPDATE 都不会产生结果集description自然是None。另外如果用了cursor.execute但没真正执行比如被异常中断也会是None。排查顺序先确认 SQL 是 SELECT再确认execute没有抛异常最后确认description在fetchall之前取。6.2 列名顺序与 SELECT 不一致正常情况下cursor.description的顺序与 SELECT 子句一致。如果发现不一致检查是否用了SELECT *加JOIN不同表的同名字段可能被覆盖。建议显式列出字段避免*。6.3 别名解析拿到的是原字段名这说明 SQL 里没有用AS或者用了但被优化器改写。MySQL 在某些版本下对别名处理有差异建议统一用AS显式声明。如果还是拿到原字段名检查连接参数里是否开了某些兼容模式。6.4 字典映射时 key 重复如果 SELECT 里有两个同名列比如a.id和b.iddict(zip(...))会保留最后一个前面的被覆盖。解决办法是给其中一个起别名或者在映射前做去重校验。def check_duplicate_columns(columns: list[str]): seen set() dupes [] for c in columns: if c in seen: dupes.append(c) seen.add(c) if dupes: raise ValueError(f存在重复列名: {dupes})6.5 中文列名编码问题cursor.description返回的列名是 str但如果数据库连接 charset 设错中文别名可能变成乱码。确保charsetutf8mb4并且数据库、表、字段的字符集一致。6.6 连接超时导致 description 取不到connect_timeout和read_timeout设太短查询还没返回就断了。生产环境建议read_timeout不低于 30 秒复杂查询可以设到 120 秒。6.7 TaoToken 调用返回 401检查settings.json里的api_key是否被环境变量覆盖成了空值。另外确认api_base没有多余斜杠正确写法是https://taotoken.net/api拼接路径时用/v1/chat/completions。排障和接入相关的入口再放一次API Keys 在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。验证模型语义映射走模型对话 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 长期编码和 Agent 场景用 Coding Plan https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 。最后补一个实用技巧把get_columns和rows_to_dicts_safe抽成独立模块后任何查询都能一行代码拿到字典列表。我试过在十几个数据同步脚本里复用这套封装列名错位的问题基本没再出现过。如果你用的是 SQLAlchemyresult.keys()也能拿到列名但 pymysql 原生方案更轻适合不想引入 ORM 的场景。