1. 为什么“读表头”这件小事总在项目里翻车
Python 连 MySQL 读表头,听起来就是一行cursor.description的事,但真到项目里,它经常变成一连串小坑的集合。比如做数据导出时,你希望 Excel 第一行是中文列名,结果拿到的是(('id', ...), ('user_name', ...))这种元组套元组;做动态建表时,你想根据源表字段自动生成CREATE TABLE,却发现字段类型、是否为空、默认值全藏在description的不同下标里;做字段校验时,上游改了列名,下游脚本静默跑完,直到报表对不上数才被发现。
这些场景的共同点是:表头不只是“名字”,它是元数据。cursor.description返回的每一项通常包含 7 个元素,最常用的是第 0 位(列名)和第 1 位(类型码),但很多人只取了第 0 位就收工,后面做类型映射时又得重新查一遍。更麻烦的是连接配置散落在各个脚本里,host、port、user、password 硬编码,换一套环境就要全局搜索替换。
所以这篇不打算只给你一段“能跑”的代码,而是把两件事一起解决:一是用config.toml把数据库连接和统一 Key 通道收拢成配置骨架,二是把cursor.description的用法讲透,包括怎么验证、怎么排错。你跟着配一遍,后面再遇到“读表头”的需求,直接改配置就能复用。
适合谁看:正在写数据导出脚本的 Python 初学者、需要动态生成 SQL 的后端同学、以及想把零散连接配置统一管理的运维/数据工程同学。核心检索词就三个:python、mysql、表头,全文围绕它们展开。
2. TaoToken 统一 Key 与 API 通道的前置准备
在讲数据库之前,先说清楚为什么这里要引入 TaoToken。很多同学的 Python 脚本里,除了连 MySQL,还会调用大模型做字段注释生成、SQL 补全、数据清洗规则推断。如果每个脚本都单独配一套 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 (这个不加 UTM)。你需要先去控制台创建一个 API Key,然后把它写进config.toml,而不是硬编码在 Python 文件里。
具体操作路径:打开控制台页面 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ,登录后进入 API Keys 管理页 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite ,点新建,复制生成的 Key。这个 Key 就是后面config.toml里[llm]段的api_key值。
注意:Key 只显示一次,复制后先存到密码管理器或本地
.env,不要直接提交到 Git。配置文件里建议用占位符,运行时再注入。
如果你只是想先验证模型通道是否通,可以打开模型对话页 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_content=model-chat&utm_campaign=rewrite 发一条消息,确认 Key 有效。这一步不是必须,但能帮你排除“到底是 Key 错还是代码错”的干扰。长期做编码和 Agent 的同学,可以了解 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ,它更适合高频调用场景。
接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ,里面有完整的请求格式和参数说明。Claude Code 相关配置参考 https://taotoken.net/claudecode-anthropic?utm_source=taotoken_aicg_blog_end&utm_content=claudecode&utm_campaign=rewrite 。这些链接先记着,后面 CTA 会分流。
3. config.toml 配置骨架与 Python 读取代码
这一节是全文的核心,给你一份可以直接复制的config.toml骨架,包含 MySQL 连接段和 TaoToken 段。然后写一个load_config函数把它读进来,再写get_table_header函数用cursor.description拿表头。
先看配置文件。放在项目根目录,命名config.toml:
# config.toml [mysql] host = "127.0.0.1" port = 3306 user = "your_user" password = "your_password" database = "your_db" charset = "utf8mb4" [mysql.pool] min_size = 1 max_size = 5 timeout = 10 [llm] base_url = "https://taotoken.net/api" api_key = "sk-替换成你的Key" model = "gpt-4o-mini" timeout = 30 [app] default_table = "orders" header_cache_ttl = 300这里有几个设计点值得说明。[mysql]段放基础连接信息,[mysql.pool]放连接池参数,方便后面换成DBUtils或SQLAlchemy时直接读。[llm]段的base_url固定指向 TaoToken 的 API 入口,api_key用占位符,实际运行时从环境变量覆盖。[app]段放业务默认值,比如默认查哪张表、表头缓存多久。
Python 3.11 之后标准库自带tomllib,可以直接读 TOML。如果你用的是 3.10 及以下,装tomli即可。下面是读取和连接代码:
# db_header.py import tomllib from pathlib import Path import pymysql from pymysql.cursors import DictCursor def load_config(path: str = "config.toml") -> dict: with open(Path(path), "rb") as f: return tomllib.load(f) def get_connection(cfg: dict): m = cfg["mysql"] return pymysql.connect( host=m["host"], port=m["port"], user=m["user"], password=m["password"], database=m["database"], charset=m.get("charset", "utf8mb4"), cursorclass=DictCursor, autocommit=True, ) def get_table_header(table: str, cfg: dict) -> list[str]: conn = get_connection(cfg) try: with conn.cursor() as cur: cur.execute(f"SELECT * FROM `{table}` LIMIT 0") return [col[0] for col in cur.description] finally: conn.close() if __name__ == "__main__": cfg = load_config() header = get_table_header(cfg["app"]["default_table"], cfg) print("表头列名:", header) print("列数:", len(header))关键点在LIMIT 0。很多人习惯SELECT * FROM table然后fetchall,数据量大时直接把内存打满。加LIMIT 0后,MySQL 只返回结果集的元数据,不返回任何行,cursor.description依然完整。这是读表头最省资源的方式,实测在千万级表上也是毫秒级返回。
cursor.description的结构是 7 元组,下标含义如下表:
| 下标 | 含义 | 示例 |
|---|---|---|
| 0 | 列名 name | user_name |
| 1 | 类型码 type_code | 253(VAR_STRING) |
| 2 | 显示长度 display_size | 255 |
| 3 | 内部长度 internal_size | 1020 |
| 4 | 精度 precision | None |
| 5 | 小数位 scale | None |
| 6 | 是否可为空 null_ok | True |
如果你要做动态建表,光有列名不够,还得把类型码映射成 MySQL 类型。下面这个映射函数可以直接用:
import pymysql.constants.FIELD_TYPE as FT TYPE_MAP = { FT.DECIMAL: "DECIMAL", FT.TINY: "TINYINT", FT.SHORT: "SMALLINT", FT.LONG: "INT", FT.FLOAT: "FLOAT", FT.DOUBLE: "DOUBLE", FT.NULL: "NULL", FT.TIMESTAMP: "TIMESTAMP", FT.LONGLONG: "BIGINT", FT.INT24: "MEDIUMINT", FT.DATE: "DATE", FT.TIME: "TIME", FT.DATETIME: "DATETIME", FT.YEAR: "YEAR", FT.NEWDATE: "DATE", FT.VARCHAR: "VARCHAR", FT.BIT: "BIT", FT.JSON: "JSON", FT.NEWDECIMAL: "DECIMAL", FT.ENUM: "ENUM", FT.SET: "SET", FT.TINY_BLOB: "TINYBLOB", FT.MEDIUM_BLOB: "MEDIUMBLOB", FT.LONG_BLOB: "LONGBLOB", FT.BLOB: "BLOB", FT.VAR_STRING: "VARCHAR", FT.STRING: "CHAR", FT.GEOMETRY: "GEOMETRY", } def describe_columns(table: str, cfg: dict) -> list[dict]: conn = get_connection(cfg) try: with conn.cursor() as cur: cur.execute(f"SELECT * FROM `{table}` LIMIT 0") result = [] for col in cur.description: result.append({ "name": col[0], "type": TYPE_MAP.get(col[1], "TEXT"), "nullable": col[6], }) return result finally: conn.close()这样你拿到的就不只是表头名字,而是一份可用来拼CREATE TABLE的字段描述。注意LIMIT 0对视图、临时表同样有效,但对SHOW COLUMNS这类语句不适用,那种场景直接用SHOW COLUMNS FROM table更直接。
4. 验证请求与成功结果检查
配置写完,得验证。分两步:先验证 MySQL 表头读取,再验证 TaoToken 通道。
第一步,跑python db_header.py。如果config.toml里default_table填的是真实存在的表,你会看到类似输出:
表头列名: ['id', 'user_name', 'amount', 'created_at'] 列数: 4如果表不存在,会抛pymysql.err.ProgrammingError: (1146, "Table 'your_db.orders' doesn't exist")。这时候先确认database配置和表名拼写,别急着改代码。
第二步,验证 TaoToken 通道。写一个最小请求脚本:
import tomllib, json, urllib.request def load_config(path="config.toml"): with open(path, "rb") as f: return tomllib.load(f) def ping_llm(cfg): llm = cfg["llm"] payload = json.dumps({ "model": llm["model"], "messages": [{"role": "user", "content": "回复 OK 两个字母"}], "max_tokens": 10, }).encode() req = urllib.request.Request( f"{llm['base_url']}/v1/chat/completions", data=payload, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {llm['api_key']}", }, ) with urllib.request.urlopen(req, timeout=llm["timeout"]) as resp: body = json.loads(resp.read()) print("模型返回:", body["choices"][0]["message"]["content"]) if __name__ == "__main__": ping_llm(load_config())运行后如果打印模型返回: OK,说明 Key 和通道都正常。如果返回 401,检查api_key是否复制完整;返回 404,检查base_url是否写成了https://taotoken.net/api而不是带路径的地址。
第三步,把两者串起来做一个真实场景:读表头后让模型生成字段中文注释。代码片段如下:
def gen_comment(header: list[str], cfg: dict) -> str: llm = cfg["llm"] prompt = f"为这些数据库字段生成简短中文注释,每行一个,格式:字段名: 注释\n" + "\n".join(header) payload = json.dumps({ "model": llm["model"], "messages": [{"role": "user", "content": prompt}], "max_tokens": 500, }).encode() req = urllib.request.Request( f"{llm['base_url']}/v1/chat/completions", data=payload, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {llm['api_key']}", }, ) with urllib.request.urlopen(req, timeout=llm["timeout"]) as resp: return json.loads(resp.read())["choices"][0]["message"]["content"]跑通后你会得到类似id: 主键、user_name: 用户名的输出。这一步验证的是“表头读取 + 模型调用”的完整链路,也是很多数据导出脚本里真正需要的组合能力。
5. 本篇常见错误排查
这一节按报错信息组织,你遇到哪个查哪个。
报错一:ModuleNotFoundError: No module named 'tomllib'
Python 3.11 以下没有tomllib。解决办法是pip install tomli,然后把import tomllib改成import tomli as tomllib。注意tomli只读不写,读配置够用。
报错二:pymysql.err.OperationalError: (2003, "Can't connect to MySQL server")
先确认 MySQL 服务在跑,再确认host和port。如果你在容器里跑 Python,127.0.0.1指向容器自身,应该改成宿主机地址或服务名。config.toml里host不要带http://前缀,只写 IP 或域名。
报错三:cursor.description返回None
只有执行了SELECT类语句后description才有值。如果你执行的是INSERT、UPDATE、CREATE,description就是None。读表头必须用SELECT ... LIMIT 0,别用SHOW或DESCRIBE混着来。
报错四:表头列名是 bytes 而不是 str
这是charset没配对。config.toml里charset = "utf8mb4",连接时传给pymysql.connect。如果还是 bytes,检查 MySQL 服务端字符集和表字符集是否一致,用SHOW CREATE TABLE确认。
报错五:TaoToken 返回 401 或 403
先确认api_key没有多余空格,再确认请求头是Authorization: Bearer sk-xxx。如果 Key 是在控制台新建的,确认没有过期或被禁用。排障时建议先打开 API Keys 页面 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 核对 Key 状态,再对照接入文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 检查请求格式。
报错六:LIMIT 0在视图上返回空 description
某些 MySQL 版本对视图加LIMIT 0时元数据不完整。这种情况改用SELECT * FROM view_name LIMIT 1,然后不取数据只取description,或者直接用information_schema.columns查询。后者更稳:
SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE FROM information_schema.columns WHERE table_schema = %s AND table_name = %s ORDER BY ordinal_position;把这条 SQL 的参数用config.toml里的database和表名传入,同样能拿到表头和类型,而且不依赖cursor.description的行为差异。
报错七:配置文件路径找不到
load_config默认读当前工作目录的config.toml。如果你在子目录跑脚本,用绝对路径或Path(__file__).parent / "config.toml"。别用相对路径../config.toml,换 IDE 运行配置就容易断。
6. 把配置和 Key 收拢后的下一步
走到这里,你已经有了三样东西:一份可复用的config.toml骨架、一个用cursor.description读表头的函数、一条验证过的 TaoToken 通道。接下来怎么用,取决于你的场景。
如果你在做数据导出,把get_table_header的返回值直接作为 DataFrame 的columns,配合pandas.read_sql就能生成带正确表头的 Excel。如果你在做动态建表,用describe_columns拿到字段名和类型后拼CREATE TABLE,注意给字符串类型补上长度,VARCHAR不带长度在严格模式下会报错。如果你在做字段校验,把表头列表和预期列表做集合差,上游改列名时第一时间告警。
关于 Key 和通道的后续使用,按场景分流:排障和接入问题看 API Keys 和接入文档;想先验证模型效果,去模型对话页发几条消息;长期做编码和 Agent 的,了解 Coding Plan 会更省心。链接都在前面给过,这里不重复贴。
最后留一个实用技巧:config.toml里的api_key不要写死,用环境变量覆盖。在load_config里加一行cfg["llm"]["api_key"] = os.environ.get("TAOTOKEN_API_KEY", cfg["llm"]["api_key"]),这样本地开发和 CI 环境可以用不同的 Key,配置文件本身可以安全提交。表头缓存也可以加,header_cache_ttl那个字段就是留给functools.lru_cache或 Redis 的,表结构不常变的话,缓存能省掉大量重复查询。