psycopg2-binary 全面教程：常用 API 串联与实战指南（TaoToken 统一 Key 接入版）

1. psycopg2-binary 到底解决什么问题从 connect 到连接池的完整链路psycopg2-binary 是 Python 操作 PostgreSQL 最常用的驱动库它是 psycopg2 的预编译二进制版本装完就能用不需要本地再装 libpq 开发包和编译器。如果你写过pip install psycopg2结果卡在pg_config executable not found换成psycopg2-binary基本就绕过了这个坑。它适合谁适合做数据脚本、后端服务、ETL 任务、测试库造数以及任何需要把 Python 和 PostgreSQL 串起来的场景。这篇教程不打算只讲单个 API而是把connect→cursor→execute→commit/rollback→fetch→ 连接池复用 → 异常重试这条链路完整跑一遍。同时我会把模型调用这一侧的 Key 管理也统一进来用 TaoToken 的统一 Key 接入方式把数据库脚本里可能用到的模型能力比如生成测试数据、解释 SQL 报错也走同一套凭证避免项目里散落一堆 Key。你可以在 https://taotoken.net/api 找到 API 入口模型对话调试在 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chat 需要长期跑编码或 Agent 任务可以看 Coding Planhttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding_plan 。先说清楚 psycopg2 的事务模型这是新手最容易翻车的地方。psycopg2 默认关闭自动提交也就是说你执行完INSERT之后如果不调用conn.commit()数据不会真正落库一旦连接关闭未提交的修改会被回滚。很多人第一次写完插入代码去数据库里查发现没数据八成就是忘了 commit。反过来如果中途报错你必须显式conn.rollback()否则这个连接会一直停留在「事务失败」状态后续所有 SQL 都会报current transaction is aborted, commands ignored until end of transaction block。理解这一点后面所有 API 才有意义。环境准备部分Python 建议 3.8 以上PostgreSQL 服务本地或远程都行默认端口 5432。安装命令直接复制# 安装最新版 pip install psycopg2-binary # 安装指定版本2.9.x 适配 PostgreSQL 12 及以上 pip install psycopg2-binary2.9.9 # 如果你用连接池 重试建议一起装上 pip install tenacity前置条件就三条PostgreSQL 已启动、有可访问的库和账号密码、目标端口没被防火墙拦。远程库记得在连接参数里加connect_timeout否则网络不通时脚本会卡很久。下面这张表是我常用的连接参数对照先记住后面配置片段会反复用到参数说明默认值host数据库地址localhostport端口5432dbname / database目标库名当前系统用户user用户名当前系统用户password密码无sslmodeSSL 模式如 requiredisableconnect_timeout连接超时秒数无把这一节的核心记住一句话psycopg2-binary 负责连接和事务连接池负责复用重试负责兜底三者串起来才是生产可用的链路。下一节先把 TaoToken 的 Key 和 Base URL 配好让模型侧和数据库侧用同一套凭证管理思路。2. TaoToken 统一 Key 前置配置Base URL、API Key 与 Model ID 三件套在写数据库脚本之前先把模型侧的凭证统一掉。原因很现实一个项目里如果数据库密码、模型 Key、各种第三方 Token 各写各的迁移环境时必然漏。TaoToken 提供统一 Key 接入你只需要记住三件套——Base URL、API Key、Model ID。Base URL 固定用https://taotoken.net/api注意这个地址不带任何查询参数API Key 在控制台创建地址是 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsole 创建后复制保存页面只显示一次。如果你用 Claude Code 这类编码工具接入文档在 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc API Keys 管理页在 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi_keys 。Claude Code 的 Anthropic 兼容入口单独放在 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentclaude_code_anthropic 需要的话直接照文档填。我习惯把凭证放环境变量不写进代码。Linux/macOS 下export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export PG_DSNpostgresql://postgres:123456localhost:5432/test_dbWindows PowerShell$env:TAOTOKEN_API_KEYsk-你的Key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api $env:PG_DSNpostgresql://postgres:123456localhost:5432/test_db这里有个细节值得展开为什么数据库连接串也建议放环境变量因为 psycopg2 支持 DSN 字符串传参把 host、port、dbname、user、password 拼成一条 URL代码里只读一个变量切换测试库和生产库时改环境变量就行不用动代码。DSN 格式是postgresql://用户:密码主机:端口/库名特殊字符记得 URL 编码。如果你用 Cline 或带 MCP 的工具配置通常是一个 JSON 文件路径按工具文档来。下面给一个通用的 settings 片段结构字段名以你实际工具为准但三件套的位置是一致的{ mcpServers: { taotoken: { command: npx, args: [-y, your-mcp-server], env: { BASE_URL: https://taotoken.net/api, API_KEY: sk-你的Key, MODEL_ID: claude-sonnet-4-5 } } } }注意 Base URL 写https://taotoken.net/api不要自己加/v1之类的后缀具体路径由 SDK 或工具拼接。Model ID 按你实际要用的模型填别照抄示例。Codex 用户如果走auth.json结构类似把 Base URL、Key、Model ID 三个字段填全即可缺一个都会在请求时报鉴权或模型不存在。配完之后先别急着写数据库代码用一段最小请求验证 Key 是否可用。这一步能帮你把「Key 问题」和「数据库问题」提前分开否则后面报错你分不清是哪一侧。验证脚本import os import requests base_url os.environ[TAOTOKEN_BASE_URL] api_key os.environ[TAOTOKEN_API_KEY] resp requests.post( f{base_url}/v1/messages, headers{ x-api-key: api_key, anthropic-version: 2023-06-01, content-type: application/json, }, json{ model: claude-sonnet-4-5, max_tokens: 64, messages: [{role: user, content: 只回复两个字连通}], }, timeout30, ) print(resp.status_code) print(resp.text[:300])返回 200 且能看到内容说明 Key 和 Base URL 没问题。如果返回 401先检查 Key 有没有复制完整、有没有多余空格如果返回 404检查 Base URL 是不是写成了带路径的形式。这一步过了再进入数据库配置排障时思路会清晰很多。3. 可复制配置连接串模板、连接池片段与 requirements这一节全是能直接抄的配置。先给 requirements把依赖固定下来避免不同机器装出不同版本psycopg2-binary2.9.9 tenacity8.2.3连接串模板我准备了两套一套是 DSN 字符串一套是关键字参数你按习惯选。DSN 适合放环境变量关键字参数适合在代码里显式写清楚import os import psycopg2 # 方式一DSN 字符串从环境变量读取 DSN os.environ.get( PG_DSN, postgresql://postgres:123456localhost:5432/test_db, ) def get_conn_by_dsn(): return psycopg2.connect(DSN, connect_timeout10) # 方式二关键字参数适合配置分离 DB_CONFIG { dbname: os.environ.get(PG_DB, test_db), user: os.environ.get(PG_USER, postgres), password: os.environ.get(PG_PASSWORD, 123456), host: os.environ.get(PG_HOST, localhost), port: os.environ.get(PG_PORT, 5432), connect_timeout: 10, } def get_conn_by_kwargs(): return psycopg2.connect(**DB_CONFIG)连接池是生产必备频繁 connect/close 会消耗大量资源。psycopg2 自带pool模块最常用的是SimpleConnectionPool和ThreadedConnectionPool。单线程脚本用前者多线程服务用后者。下面这段可以直接用from psycopg2 import pool connection_pool pool.ThreadedConnectionPool( minconn2, maxconn10, dbnametest_db, userpostgres, password123456, hostlocalhost, port5432, connect_timeout10, ) def with_pool(): conn connection_pool.getconn() try: with conn.cursor() as cur: cur.execute(SELECT 1;) print(pool ok:, cur.fetchone()) conn.commit() except Exception: conn.rollback() raise finally: # 归还连接不是关闭 connection_pool.putconn(conn)这里有个坑要提醒putconn归还连接时如果这个连接处于失败事务状态池子里下次取出来还是坏的。稳妥做法是在except里先rollback再归还或者归还时传closeTrue直接丢弃这个连接。我一般用前者因为回滚成本低。再给一个带重试的封装用 tenacity 做指数退避。数据库偶发连接抖动、网络瞬断时重试能救回不少任务from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import psycopg2 retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max8), retryretry_if_exception_type(psycopg2.OperationalError), reraiseTrue, ) def run_with_retry(sql, paramsNone): conn connection_pool.getconn() try: with conn.cursor() as cur: cur.execute(sql, params) if cur.description: rows cur.fetchall() else: rows None conn.commit() return rows except Exception: conn.rollback() raise finally: connection_pool.putconn(conn)注意重试只包OperationalError也就是连接层面的错误。像唯一约束冲突这种IntegrityError不该重试重试多少次都一样反而拖慢流程。这个区分很重要很多人一把梭全重试结果业务错误被反复执行。最后把连接池的关闭也写清楚程序退出时调用connection_pool.closeall()否则连接会挂在那直到超时。如果你用 FastAPI 或 Flask放在应用 shutdown 钩子里import atexit atexit.register(connection_pool.closeall)配置到这一步连接串、连接池、重试三件套齐了。下一节用 SELECT/INSERT/UPDATE 实际验证连通性和事务回滚把 API 串起来跑通。4. 验证请求与成功结果SELECT/INSERT/UPDATE 串联与事务回滚实测先建一张测试表把字段设计得能覆盖常见类型自增主键、字符串、整数、唯一约束、时间戳、JSONB。建表语句CREATE_TABLE_SQL CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, name VARCHAR(50) NOT NULL, age INT, email VARCHAR(100) UNIQUE, create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, info JSONB ); def create_table(conn): with conn.cursor() as cur: cur.execute(CREATE_TABLE_SQL) conn.commit() print(建表完成)注意with conn.cursor() as cur这种写法退出 with 块时游标自动关闭比手动cur.close()省心也不会漏。但连接不会自动关连接还是要自己管。插入用参数化查询占位符是%s不是 Python 的{}也不是%d。这是 psycopg2 的硬性规定写错了会报TypeError或 SQL 语法错误。批量插入用execute_batch比循环execute快很多from psycopg2.extras import execute_batch, Json INSERT_SQL INSERT INTO users (name, age, email, info) VALUES (%s, %s, %s, %s) ON CONFLICT (email) DO NOTHING; def insert_users(conn, rows): with conn.cursor() as cur: execute_batch(cur, INSERT_SQL, rows, page_size100) conn.commit() print(f插入 {len(rows)} 条) rows [ (张三, 25, zhangsanexample.com, Json({hobby: [篮球]})), (李四, 28, lisiexample.com, Json({hobby: [阅读]})), (王五, 30, wangwuexample.com, Json({hobby: [跑步]})), ]ON CONFLICT (email) DO NOTHING保证重复邮箱不会报错适合反复跑脚本的场景。JSONB 字段用Json()包一层psycopg2 会自动序列化查询出来自动反序列化成 dict。查询用fetchall但生产环境大结果集别用改fetchmany分批。这里演示三种取数方式def query_users(conn, age_min0): sql SELECT id, name, age, email, info FROM users WHERE age %s ORDER BY id; with conn.cursor() as cur: cur.execute(sql, (age_min,)) print(rowcount:, cur.rowcount) one cur.fetchone() print(fetchone:, one) rest cur.fetchmany(2) print(fetchmany:, rest) left cur.fetchall() print(fetchall:, left)rowcount在 SELECT 时是匹配行数在 INSERT/UPDATE/DELETE 时是影响行数判断更新是否命中很有用。更新操作def update_age(conn, email, new_age): sql UPDATE users SET age %s WHERE email %s; with conn.cursor() as cur: cur.execute(sql, (new_age, email)) affected cur.rowcount conn.commit() print(f更新影响 {affected} 行)现在重点来了事务回滚实测。故意在同一个事务里先插一条合法数据再插一条重复邮箱触发唯一约束错误看第一条会不会被回滚def test_rollback(conn): try: with conn.cursor() as cur: cur.execute( INSERT INTO users (name, age, email) VALUES (%s, %s, %s);, (临时用户, 99, tempexample.com), ) # 下面这条会触发唯一约束冲突 cur.execute( INSERT INTO users (name, age, email) VALUES (%s, %s, %s);, (重复用户, 88, zhangsanexample.com), ) conn.commit() print(提交成功) except Exception as e: conn.rollback() print(触发回滚:, type(e).__name__, e)跑完之后再查tempexample.com你会发现查不到。这就是事务的原子性要么全成功要么全回滚。如果你不写conn.rollback()这个连接后续所有操作都会报current transaction is aborted必须回滚才能继续用。我踩过的坑就是早期在 except 里只 print 不 rollback结果循环里第二条数据开始全部失败排查了半天。成功结果长这样你可以对照自己的输出建表完成 插入 3 条 rowcount: 3 fetchone: (1, 张三, 25, zhangsanexample.com, {hobby: [篮球]}) fetchmany: [(2, 李四, 28, lisiexample.com, {hobby: [阅读]})] fetchall: [(3, 王五, 30, wangwuexample.com, {hobby: [跑步]})] 更新影响 1 行 触发回滚: UniqueViolation duplicate key value violates unique constraint users_email_key看到UniqueViolation且临时用户没入库说明事务回滚生效。到这一步connect、cursor、execute、commit、rollback、fetch 全链路就验证完了。下一节集中处理常见报错。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 对照排错的核心思路是分层先确认是模型侧还是数据库侧再往下钻。下面按真实报错逐条对照。401 Unauthorized。这个基本出在模型侧。原因通常是 Key 没读到、Key 复制不全、或者请求头字段名写错。Anthropic 兼容接口用x-api-keyOpenAI 兼容接口用Authorization: Bearer。先打印一下环境变量确认非空import os print(repr(os.environ.get(TAOTOKEN_API_KEY)))如果打印出None或带引号说明环境变量没生效。另外检查 Base URL 是不是https://taotoken.net/api多写路径会导致鉴权失败。Key 管理页在 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi_keys 重新生成一个对比测试。local proxy failed。这个报错一般出现在工具或 SDK 层提示本地代理连接失败。先检查你的运行环境有没有配置HTTP_PROXY、HTTPS_PROXY环境变量如果配了但代理服务没起来就会报这个。清掉环境变量再试unset HTTP_PROXY HTTPS_PROXY ALL_PROXY同时确认 Base URL 是直连可达的用 curl 测一下curl -sS -o /dev/null -w %{http_code}\n https://taotoken.net/api返回 200 或 401 都说明网络通返回 000 才是网络问题。reading choices 相关报错。这类错误通常出现在解析响应时比如KeyError: choices或reading choices。原因是接口返回结构和你代码里取字段的路径不匹配。OpenAI 兼容格式是resp[choices][0][message][content]Anthropic 格式是resp[content][0][text]。先打印完整响应再取字段data resp.json() print(data.keys())看清楚顶层有哪些 key再决定怎么取。别硬编码路径加个兜底判断。OAuth 相关报错。如果你用 Claude Code 或 Codex 这类工具报 OAuth 失败通常是认证方式选错了。走 API Key 接入时不需要 OAuth 流程检查工具配置里是不是还留着 OAuth 的字段。Claude Code 的 Anthropic 兼容配置参考 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentclaude_code_anthropic 把 Base URL、Key、Model ID 三件套填全OAuth 字段清空。数据库侧的常见错也列一下。OperationalError: could not connect to server检查 host/port/服务是否启动password authentication failed检查密码和pg_hba.confcurrent transaction is aborted就是前面说的忘了 rollbackrelation users does not exist检查建表有没有 commit或者 search_path 不对。还有一个隐蔽的连接池取出的连接是坏的。表现是第一次请求正常后面偶发server closed the connection unexpectedly。解决办法是归还前 rollback或者用connection_pool.putconn(conn, closeTrue)丢弃。我一般加个健康检查取出来先SELECT 1def get_healthy_conn(): conn connection_pool.getconn() try: with conn.cursor() as cur: cur.execute(SELECT 1;) return conn except Exception: connection_pool.putconn(conn, closeTrue) return connection_pool.getconn()排错时记住一个原则先隔离再定位。模型侧和数据库侧分开测各自能通之后再串起来问题范围会小很多。6. 长期跑数据库脚本与 Agent 任务把 Key 和连接池一起管起来如果你只是偶尔跑个脚本前面五节够用了。但如果你要长期跑 ETL、定时任务或者用 Agent 自动生成测试数据、自动排查 SQL 报错那凭证管理和连接管理就得当成基础设施来做。我现在的做法是数据库连接池在应用启动时初始化一次模型侧统一走 TaoToken 的 Key两边都从环境变量读代码里不出现任何明文。模型侧如果调用频繁建议单独封装一个客户端把 Base URL、Key、Model ID 固定住业务代码只传 prompt。这样换模型或换 Key 只改一处。需要长期编码或 Agent 任务的可以看 Coding Planhttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding_plan 模型对话调试用 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chat 接入文档在 https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc 。连接池这块长期运行的服务要监控池子使用率。ThreadedConnectionPool没有直接暴露活跃连接数但你可以自己计数getconn 时加一putconn 时减一超过阈值告警。另外设置maxconn别太大PostgreSQL 默认max_connections是 100多个服务一起抢容易打满。一般单服务 10 到 20 够用。重试策略也要按场景调。读操作可以多试几次写操作要小心幂等。比如 INSERT 重试可能插重复数据除非你有唯一约束兜底。UPDATE 重试一般安全因为幂等。DELETE 同理。我通常给读操作配 3 次重试写操作配 1 次失败就进死信队列人工处理。最后给一个把数据库和模型串起来的小例子用模型生成测试数据再批量写入 PostgreSQL。这样你能看到两边怎么协作import json import os import psycopg2 import requests from psycopg2.extras import execute_batch, Json def gen_users(n5): resp requests.post( f{os.environ[TAOTOKEN_BASE_URL]}/v1/messages, headers{ x-api-key: os.environ[TAOTOKEN_API_KEY], anthropic-version: 2023-06-01, content-type: application/json, }, json{ model: claude-sonnet-4-5, max_tokens: 512, messages: [{ role: user, content: f生成 {n} 条用户测试数据JSON 数组字段 name/age/email只输出 JSON。, }], }, timeout60, ) text resp.json()[content][0][text] return json.loads(text) def save_users(conn, users): sql INSERT INTO users (name, age, email, info) VALUES (%s, %s, %s, %s) ON CONFLICT (email) DO NOTHING; rows [(u[name], u[age], u[email], Json({source: llm})) for u in users] with conn.cursor() as cur: execute_batch(cur, sql, rows, page_size100) conn.commit() print(f写入 {len(rows)} 条) if __name__ __main__: conn psycopg2.connect(os.environ[PG_DSN], connect_timeout10) try: users gen_users(5) save_users(conn, users) finally: conn.close()这段代码把模型生成、JSON 解析、批量写入、冲突忽略串成一条线。跑通之后你可以把它改成定时任务或者接进 Agent 流程。核心还是那句话连接池管复用重试管兜底统一 Key 管凭证三者各司其职脚本才能长期稳定跑下去。