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_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_content=api_keys&utm_campaign=rewrite

模型对话入口（验证字段语义映射时用）：https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite

接入文档：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite

长期做编码和 Agent 的话，Coding Plan 更合适：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite

Console 控制台：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite

ClaudeCodeAnthropic 入口：https://taotoken.net/claude-code-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claude_code&utm_campaign=rewrite

注意：TaoToken 是统一 API 通道，不是数据库代理。pymysql 仍然直连你的 MySQL，TaoToken 负责的是查询结果后续的模型调用与凭证管理。

3. 可复制配置：config.toml 与 settings.json

先把配置骨架搭好。我习惯把数据库连接和 API 通道分开管理，数据库走config.toml，API 凭证走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 = 60

fetch_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", encoding="utf-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/api

4. 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, rows

cursor.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 result

zip在列名和行长度不一致时会静默截断，生产环境建议用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( host=self.db_cfg["host"], port=self.db_cfg["port"], user=self.db_cfg["user"], password=self.db_cfg["password"], database=self.db_cfg["database"], charset=self.db_cfg["charset"], connect_timeout=self.db_cfg["connect_timeout"], read_timeout=self.db_cfg["read_timeout"], cursorclass=pymysql.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_ascii=False) ) 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", data=payload, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {cfg['api_key']}", }, method="POST", ) with urllib.request.urlopen(req, timeout=cfg["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.id），dict(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 设错，中文别名可能变成乱码。确保charset="utf8mb4"，并且数据库、表、字段的字符集一致。

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_source=taotoken_aicg_blog_end&utm_content=api_keys&utm_campaign=rewrite ，接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。验证模型语义映射走模型对话 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite ，长期编码和 Agent 场景用 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。

最后补一个实用技巧：把get_columns和rows_to_dicts_safe抽成独立模块后，任何查询都能一行代码拿到字典列表。我试过在十几个数据同步脚本里复用这套封装，列名错位的问题基本没再出现过。如果你用的是 SQLAlchemy，result.keys()也能拿到列名，但 pymysql 原生方案更轻，适合不想引入 ORM 的场景。