用Python查询数据库返回字段名和值组成的字典类型：TaoToken统一Key通道下的游标处理实践-育师

1. 从一次真实查询说起：为什么默认游标返回的不是字典

很多刚接触 Python 数据库操作的朋友，第一次写查询代码时都会遇到同一个困惑：明明 SQL 里select *拿到了所有字段，可fetchall()打印出来却是一堆没有名字的元组，比如(1, '张三', 28)。你根本不知道哪个值是 id、哪个是姓名、哪个是年龄，只能靠数位置去猜。这在字段少的时候还能忍，一旦表里有十几个字段，维护起来就是灾难。

这个问题的本质在于：数据库驱动默认使用的游标（cursor）只负责把结果集按行返回，它并不关心列名。列名信息其实一直存在，只是被放在了一个叫cursor.description的属性里。description是一个元组序列，每个元素描述一列，其中第 0 个位置就是字段名。只要我们把这个字段名列表和每一行的值做一次zip，就能拼出{'id': 1, 'name': '张三', 'age': 28}这样的字典。

所以「用 Python 查询数据库返回字段名和值组成的字典类型」这件事，核心就两招：要么让驱动直接给你字典游标，要么自己拿description手动映射。前者省事，后者通用。本文会覆盖 sqlite3、MySQL、PostgreSQL 三种最常见的场景，给你可以直接复制的连接配置、字段映射函数和验证脚本，同时说明怎么用 TaoToken 的统一 Key 通道来管理多环境凭据，避免把账号密码硬编码在代码里。适合正在写数据脚本、做小工具、或者维护多套数据库环境的开发者。

我试过在同一个项目里同时连 sqlite 本地库和远程 MySQL，一开始每个文件都写一遍连接参数，改一次密码要翻五六个文件，后来统一走一个凭据通道才清爽下来。下面按步骤来。

2. 前置准备：TaoToken 统一 Key 通道与多环境凭据管理

在写查询代码之前，先把「连哪个库、用什么账号」这件事管起来。硬编码连接串的问题不只是丑，而是当你有开发、测试、生产三套环境时，密码散落在各个脚本里，改一次就漏一处。TaoToken 提供统一 Key 和 API 通道，可以把模型调用和凭据管理收敛到一个入口，配合环境变量使用，脚本里只读变量不写明文。

你需要先拿到一个可用的 Key。访问控制台创建：

控制台入口：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite
API Keys 管理：https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite

创建好 Key 之后，不要直接写进.py文件。推荐做法是放进项目根目录的.env，再用python-dotenv读取。这样.env可以加进.gitignore，团队协作时每人维护自己的一份。下面是一个.env示例，把数据库连接信息和 TaoToken 的 Key 分开管理：

# .env TAOTOKEN_API_KEY=sk-你的统一Key TAOTOKEN_BASE_URL=https://taotoken.net/api # 开发环境数据库 DEV_DB_HOST=127.0.0.1 DEV_DB_PORT=3306 DEV_DB_USER=root DEV_DB_PASSWORD=dev_pass DEV_DB_NAME=my_db # 生产环境数据库（示例，实际按需） PROD_DB_HOST=10.0.0.12 PROD_DB_PORT=3306 PROD_DB_USER=app_reader PROD_DB_PASSWORD=strong_pass PROD_DB_NAME=my_db

读取时用一个小的配置加载函数，根据APP_ENV决定用哪套前缀。这样切换环境只改一个变量，不用动代码：

import os from dotenv import load_dotenv load_dotenv() def get_db_config(env: str = None) -> dict: env = (env or os.getenv("APP_ENV", "DEV")).upper() prefix = f"{env}_DB_" return { "host": os.getenv(prefix + "HOST"), "port": int(os.getenv(prefix + "PORT", "3306")), "user": os.getenv(prefix + "USER"), "password": os.getenv(prefix + "PASSWORD"), "database": os.getenv(prefix + "NAME"), }

注意：.env文件务必加入.gitignore，不要把真实密码提交到仓库。TaoToken 的 Key 同理，只放在本地环境变量或密钥管理服务里。

如果你还想在脚本里顺带调用模型做字段注释生成、SQL 解释之类的辅助工作，可以直接复用同一个 Key，Base URL 填https://taotoken.net/api，模型 ID 按文档选择。这样数据库凭据和模型凭据都在一套体系里，排查问题时不用来回找。

3. 可复制配置：三种数据库的字典游标写法

这一节是全文的核心，直接给可运行的代码。三种数据库的差异主要在驱动和游标类的选择上，字段映射的通用逻辑是一致的。

3.1 sqlite3：用 row_factory 一步到位

sqlite3 是 Python 标准库自带的，不需要额外安装。它没有「字典游标」这个概念，但提供了row_factory，把每一行转成sqlite3.Row对象，这个对象支持按字段名索引，也能转成字典：

import sqlite3 def query_sqlite(db_path: str, sql: str, params: tuple = ()): conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row # 关键：让行支持字段名访问 cur = conn.cursor() cur.execute(sql, params) rows = cur.fetchall() result = [dict(row) for row in rows] # 转成标准字典 cur.close() conn.close() return result if __name__ == "__main__": data = query_sqlite("test.db", "select * from t_basic where tbaid = ?", (1,)) print(data)

sqlite3.Row的好处是既保留了元组的轻量，又能row["name"]这样取值。dict(row)之后就是纯字典，方便后续 JSON 序列化。

3.2 MySQL：DictCursor 直接返回字典

MySQL 用pymysql时，只要在连接时指定cursorclass=pymysql.cursors.DictCursor，fetchall()出来的就是字典列表，不需要手动 zip。这正是很多教程里提到的写法：

import pymysql from config import get_db_config def query_mysql(sql: str, params: tuple = (), env: str = None): cfg = get_db_config(env) conn = pymysql.connect( host=cfg["host"], port=cfg["port"], user=cfg["user"], password=cfg["password"], database=cfg["database"], charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, # 关键 ) try: with conn.cursor() as cur: cur.execute(sql, params) return cur.fetchall() finally: conn.close()

注意charset建议用utf8mb4，避免中文和 emoji 存储出问题。with conn.cursor()会自动关闭游标，比手动cur.close()更稳。

3.3 PostgreSQL：RealDictCursor 与 description 手动映射

PostgreSQL 用psycopg2时，可以用psycopg2.extras.RealDictCursor，效果和 MySQL 的 DictCursor 类似：

import psycopg2 import psycopg2.extras from config import get_db_config def query_pg(sql: str, params: tuple = (), env: str = None): cfg = get_db_config(env) conn = psycopg2.connect( host=cfg["host"], port=cfg["port"], user=cfg["user"], password=cfg["password"], dbname=cfg["database"], ) try: with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: cur.execute(sql, params) return [dict(r) for r in cur.fetchall()] finally: conn.close()

3.4 通用兜底：用 cursor.description 手动 zip

如果你用的驱动没有字典游标，或者想写一个不依赖具体驱动的通用函数，那就用cursor.description。它的每个元素第 0 位是列名：

def rows_to_dicts(cursor, rows): columns = [col[0] for col in cursor.description] return [dict(zip(columns, row)) for row in rows]

这个函数对 sqlite3、pymysql、psycopg2 都适用，是真正的通用写法。下面这张表帮你快速对照三种场景的选择：

数据库	驱动	字典方案	是否需手动 zip
sqlite3	标准库	`row_factory = sqlite3.Row`	否，`dict(row)`即可
MySQL	pymysql	`cursorclass=DictCursor`	否
PostgreSQL	psycopg2	`cursor_factory=RealDictCursor`	否
任意	任意	`cursor.description`+ zip	是

提示：无论用哪种方案，都要保证execute之后再读description，否则它是None。

4. 验证请求：跑通一次查询并打印字典结果

配置写好了，接下来验证。先建一张测试表，插入两条数据，然后分别用三种方式查询，确认输出都是字典。

import sqlite3 def setup_sqlite(): conn = sqlite3.connect("test.db") cur = conn.cursor() cur.execute(""" create table if not exists t_basic ( tbaid integer primary key, name text, age integer, city text ) """) cur.execute("delete from t_basic") cur.executemany( "insert into t_basic (tbaid, name, age, city) values (?, ?, ?, ?)", [(1, "张三", 28, "杭州"), (2, "李四", 34, "成都")], ) conn.commit() cur.close() conn.close() def query_sqlite(db_path, sql, params=()): conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row cur = conn.cursor() cur.execute(sql, params) rows = cur.fetchall() result = [dict(row) for row in rows] cur.close() conn.close() return result if __name__ == "__main__": setup_sqlite() data = query_sqlite("test.db", "select * from t_basic where tbaid = ?", (1,)) print("查询结果：", data) print("类型：", type(data[0])) print("字段名：", list(data[0].keys()))

运行后你应该看到类似输出：

查询结果： [{'tbaid': 1, 'name': '张三', 'age': 28, 'city': '杭州'}] 类型： <class 'dict'> 字段名： ['tbaid', 'name', 'age', 'city']

如果换成 MySQL，把query_sqlite换成第 3.2 节的query_mysql，SQL 里的占位符从?改成%s，其余逻辑不变。PostgreSQL 同理，占位符也是%s。这一步跑通，说明你的字典映射链路是通的。

如果你还想让脚本自动根据字段名生成中文注释，或者把查询结果丢给模型做摘要，可以复用 TaoToken 的模型对话能力，Base URL 用https://taotoken.net/api，Key 用前面.env里的那个。模型对话入口在这里：

模型对话：https://taotoken.net/model-chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite

这样数据库查询和模型辅助在同一个 Key 体系下，调试时只需要确认一个凭据是否有效。

5. 本篇常见错排查：401、local proxy failed 与 reading choices

写这类脚本时，报错往往不在 SQL 本身，而在连接和凭据环节。下面几个是我和身边朋友踩过的坑，对照着看。

报错一：pymysql.err.OperationalError: (1045, "Access denied for user ...")

这是数据库账号密码不对，或者该账号没有从当前主机连接的权限。先确认.env里的DEV_DB_USER和DEV_DB_PASSWORD是否和数据库实际一致。MySQL 8 默认的认证插件是caching_sha2_password，老版本 pymysql 可能不兼容，升级pip install -U pymysql通常能解决。

报错二：401 Unauthorized或invalid api key

如果你在脚本里调用了 TaoToken 的模型接口，出现 401 说明 Key 无效或没带上。检查请求头是不是Authorization: Bearer sk-xxx，以及.env里的TAOTOKEN_API_KEY有没有被正确加载。可以用一行代码验证：

import os from dotenv import load_dotenv load_dotenv() print(os.getenv("TAOTOKEN_API_KEY")[:8] + "...")

如果打印出来是None...，说明.env没被读到，检查文件路径和load_dotenv()的调用位置。

报错三：local proxy failed或连接超时

这类错误通常是网络层的问题，比如目标地址不可达、端口写错、或者本机网络策略拦截。先ping一下数据库主机，再用telnet host port确认端口通不通。如果是云数据库，检查安全组有没有放行你的出口 IP。注意不要使用任何非正规的网络中转手段，合规的网络环境是前提。

报错四：KeyError: 'reading choices'或解析响应失败

调用模型接口时，如果返回体结构和预期不一致，解析choices字段就会报 KeyError。先打印原始响应看看：

import requests, os from dotenv import load_dotenv load_dotenv() resp = requests.post( "https://taotoken.net/api/v1/chat/completions", headers={"Authorization": f"Bearer {os.getenv('TAOTOKEN_API_KEY')}"}, json={"model": "你的模型ID", "messages": [{"role": "user", "content": "hi"}]}, timeout=30, ) print(resp.status_code) print(resp.text[:500])

先确认status_code是 200，再看resp.json()里有没有choices。如果返回的是错误信息，按错误码处理，不要盲目改解析代码。

报错五：cursor.description是 None

这通常是因为在execute之前就访问了description，或者执行的是insert/update这类不返回结果集的语句。记住：只有select之后description才有值。

注意：如果你用的是 Cline MCP 或 Claude Code 这类工具去连数据库，配置里同样要写全三件套——Base URL、Key、Model ID。Base URL 填https://taotoken.net/api，Key 用统一 Key，Model ID 按文档选。缺任何一个都会连不上。