1. 为什么我建议你早点把下标取值换掉
Python 连 PostgreSQL,十有八九用的是 psycopg2。默认情况下,cursor.fetchall()返回的是元组列表,你只能靠row[0]、row[3]这种下标去拿字段。写的时候爽,维护的时候想哭:SQL 里加一个字段、调一次顺序,所有下标全乱套,尤其是select *这种写法,字段顺序完全由表结构决定,你根本控制不了。
这篇就聚焦一件事:Python 通过 psycopg2 连接 PostgreSQL 后,怎么按列名取数据。核心是cursor_factory这个参数,配上RealDictCursor或DictCursor,取数就从row[2]变成row['name'],SQL 字段顺序变了也不影响你。适合刚接触 psycopg2 的同学,也适合手里有一堆下标取值老代码、想平滑替换的人。
我会给出可直接复制的连接配置、查询代码、结果校验动作,以及几个我实际踩过的坑。你照着跑一遍,基本就能把项目里的硬编码下标换掉了。
2. 前置准备:psycopg2 安装与 TaoToken 接入配置
先说环境。psycopg2 有两个常见包名:psycopg2和psycopg2-binary。本地开发、写脚本、做验证,直接用 binary 版最省事,不用折腾编译依赖;生产环境如果对二进制分发有顾虑,再装源码版。
pip install psycopg2-binary如果你后面要接大模型来做 SQL 生成、字段解释、报错分析这类活儿,可以顺手把 TaoToken 的接入信息配好。它的 API 地址是https://taotoken.net/api,兼容常见的 OpenAI 风格调用方式,模型对话、Coding Plan、API Keys 都在控制台里管理。我一般会把 base_url 和 key 放在环境变量里,不写死在代码中:
export TAOTOKEN_API_KEY="你的key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"需要生成或管理 Key 的话,走这个入口: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 。长期写代码、跑 Agent 的话,Coding Plan 更划算:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。
数据库这边,准备一张测试表就够了。下面这段 SQL 你可以直接在 psql 或任意客户端里执行:
create table if not exists t1( id serial primary key, name text, age int, created_at timestamp default now() ); insert into t1(name, age) select 'name' || g, (g % 60) + 18 from generate_series(1, 100) as g;这张表有 4 个字段,故意让id和name不相邻,就是为了后面演示「下标取值容易错、列名取值不会错」。
3. 可复制配置:DictCursor 与 RealDictCursor 的写法
psycopg2 默认的 cursor 返回 tuple。要按列名取值,关键就在创建 cursor 时传cursor_factory。常用的有两种:
| cursor_factory | 返回类型 | 取值方式 | 特点 |
|---|---|---|---|
| 默认(不传) | tuple | row[0] | 只能下标,顺序敏感 |
DictCursor | 类字典对象 | row['name'] | 支持下标也支持列名 |
RealDictCursor | 真正的 dict | row['name'] | 纯字典,序列化友好 |
DictCursor的好处是兼容老代码,row[0]和row['name']都能用,适合渐进式替换;RealDictCursor返回的就是标准 dict,直接json.dumps不会报错,适合接口返回场景。
先看DictCursor的完整可复制版本:
# coding: utf-8 import psycopg2 import psycopg2.extras conn = psycopg2.connect( database="postgres", user="chris", password="", host="localhost", port=5432, ) cursor = conn.cursor(cursor_factory=psycopg2.extras.DictCursor) sql = "select id, name, age, created_at from t1 order by id limit 5" cursor.execute(sql) rows = cursor.fetchall() for row in rows: print(row["id"], row["name"], row["age"], row["created_at"]) cursor.close() conn.close()再看RealDictCursor,区别只在 factory 和返回类型:
import psycopg2 import psycopg2.extras conn = psycopg2.connect( database="postgres", user="chris", password="", host="localhost", port=5432, ) cursor = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cursor.execute("select id, name, age from t1 order by id limit 3") rows = cursor.fetchall() for row in rows: print(type(row), row["name"]) # RealDictCursor 返回的是 dict,可以直接序列化 import json print(json.dumps(rows, default=str, ensure_ascii=False))这里有个细节值得说:RealDictCursor返回的 dict 里,created_at是datetime对象,json.dumps默认处理不了,所以要加default=str。这也是很多人第一次用 RealDictCursor 时遇到的报错来源。
连接参数建议不要硬编码。用psycopg2.connect的 keyword 形式,或者直接传 DSN 字符串都行:
import os conn = psycopg2.connect( host=os.getenv("PGHOST", "localhost"), port=int(os.getenv("PGPORT", 5432)), dbname=os.getenv("PGDATABASE", "postgres"), user=os.getenv("PGUSER", "chris"), password=os.getenv("PGPASSWORD", ""), )4. 验证请求与成功结果:怎么确认真的按列名取到了
写完代码别急着上项目,先做三步验证。
第一步,确认返回类型。DictCursor返回的是DictRow,它继承自 list,所以isinstance(row, list)是 True,但row['name']也能用。RealDictCursor返回的是dict。你可以直接打印类型:
row = rows[0] print(type(row)) # DictCursor: <class 'psycopg2.extras.DictRow'> # RealDictCursor: <class 'dict'> print(row.keys()) # 两种都能列出列名第二步,故意打乱 SQL 字段顺序,看取值是否还正确。这是验证「按列名取值」最直接的方式:
cursor.execute("select name, id, age from t1 order by id limit 1") row = cursor.fetchone() print(row["id"], row["name"]) # 顺序变了,列名取值依然正确如果还是用row[0],这里拿到的就是 name 而不是 id,很容易出 bug。
第三步,校验字段是否存在。列名写错时,DictCursor会抛KeyError,RealDictCursor同样。你可以用in判断,或者用.get()兜底:
if "nickname" in row: print(row["nickname"]) else: print("字段不存在,走默认逻辑") # 或者 print(row.get("nickname", "未设置"))实测下来,RealDictCursor的.get()行为和普通 dict 一致,用起来最顺手。跑通后你应该能看到类似输出:
1 name1 18 2024-05-01 10:00:00 2 name2 19 2024-05-01 10:00:005. 本篇常见错排查
报错一:TypeError: tuple indices must be integers or slices, not str
这是最典型的。原因就是你用了默认 cursor,却想用列名取值。解决方式就是创建 cursor 时加cursor_factory=psycopg2.extras.DictCursor或RealDictCursor。注意 factory 是在conn.cursor()时传,不是在connect()时传,很多人第一次会传错位置。
报错二:KeyError: 'name'
列名拼错,或者 SQL 里用了别名但你还按原字段名取。比如select name as user_name from t1,这时候要取row['user_name']。另外 PostgreSQL 默认把未加引号的标识符转成小写,如果你建表时用了"Name"这种带引号的大写字段,取值时也得写row['Name'],大小写敏感。
报错三:RealDictCursor结果json.dumps失败
datetime、Decimal、UUID这些类型不能直接序列化。加default=str最省事,或者自己写一个 encoder。别去改 cursor,问题不在它。
报错四:连接后查不到数据,但表里明明有
检查是否忘了conn.commit()。psycopg2 默认开启事务,execute后的写操作不 commit 不生效。查询本身不需要 commit,但如果你在同一个连接里先插入再查询,没 commit 就查不到。另外确认search_path和表所在的 schema 一致。
报错五:DictCursor和RealDictCursor混用导致类型判断出错
有的老代码里判断isinstance(row, dict),用DictCursor时返回的是DictRow,不是 dict,判断会失败。要么统一用RealDictCursor,要么把判断改成hasattr(row, 'keys')。
6. 把列名取值接进你的项目:下一步怎么做
替换硬编码下标这件事,建议分两步走:先把新写的查询全部改成RealDictCursor,再把老代码里row[0]、row[1]逐个换成列名。换的时候顺手加个字段存在性判断,能挡掉不少线上问题。
如果你想让大模型帮你批量改写这些取值代码,或者根据表结构自动生成带列名的查询,可以走模型对话入口:https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 。需要先拿 Key 的话在这里: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/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite 。长期写代码、跑 Agent 任务,用 Coding Plan 更稳:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding_plan&utm_campaign=rewrite 。
最后留一个我常用的习惯:查询函数统一返回list[dict],内部一律用RealDictCursor,出口处做一次类型转换。这样上层业务永远拿到的是标准字典,序列化、断言、mock 都省心。