1. psycopg2 连接 PostgreSQL 的典型场景与高频坑psycopg2 是 Python 生态里操作 PostgreSQL 最成熟的驱动之一它能做什么简单说它把 Python 代码和 PostgreSQL 数据库之间的通信封装成一套稳定的 API让你用connect()拿连接、用cursor()执行 SQL、用commit()提交事务。适合谁适合需要做数据管道、后台服务、定时任务、报表导出、批量数据迁移的 Python 开发者。尤其是当你的数据量从几百行涨到几十万行时psycopg2 的连接池和批量写入能力就成了绕不开的话题。我见过太多项目一开始用psycopg2.connect()每次请求新建连接本地跑得好好的一上生产就报FATAL: sorry, too many clients already。PostgreSQL 默认max_connections是 100一个 Web 服务并发稍微上来连接数瞬间打满。还有人用executemany插十万行数据跑了十几分钟没结束最后发现是每条 INSERT 都单独发一次网络往返。这些问题的根子不在 SQL 写得对不对而在连接管理和批量策略没配对。这篇内容聚焦三个高频场景连接池怎么配、游标怎么用、批量写入怎么快。我会给出可直接复制的配置骨架包括psycopg2.pool和psycopg2.extras的用法再配合本地验证查询和性能对比动作。你跟着做一遍就能把「能跑」升级成「跑得稳、跑得快」。另外如果你在调试过程中需要快速验证模型生成的 SQL 或配置片段可以用 TaoToken 的模型对话功能做辅助校验地址在文末 CTA 里。先说一个容易被忽略的点psycopg2 的事务是自动开启的。你第一次调用cursor.execute()时事务就已经开始了之后所有 SQL 都在同一个事务里直到你显式commit()或rollback()。这意味着一个长时间运行的查询会持有锁阻塞其他写入。很多「数据库突然变慢」的工单最后都定位到某个忘了提交的只读连接上。所以连接池不仅要管连接数量还要管事务生命周期。还有一个坑是占位符。psycopg2 用%s作为参数占位符不是?也不是:name。你写WHERE id %s然后传 tuple驱动会做转义和类型适配。但如果你手写字符串拼接SQL 注入风险立刻上来。我建议所有值都走参数化包括LIMIT和OFFSET虽然它们看起来像整数但参数化能避免类型推断的边界问题。最后提一下autocommit。默认是False也就是手动事务模式。如果你做的是大量独立的 INSERT每条都 commit 会非常慢因为每次 commit 都是一次 fsync。这时候要么攒批要么临时开autocommit。但autocommitTrue下没有回滚保护适合日志类、监控类写入不适合有强一致要求的业务表。选哪种取决于你的数据能不能丢、能不能重放。2. TaoToken 前置准备API Key 与接入文档怎么拿在进入代码之前先把工具链准备好。TaoToken 在这里的角色是提供一个统一的模型调用入口方便你在写 psycopg2 代码时快速验证 SQL 逻辑、生成测试数据、或者让模型帮你审查连接池配置。它不是数据库也不替代 PostgreSQL而是一个辅助编码和调试的通道。第一步打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册并登录。登录后进入控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。在控制台左侧找到「API Keys」菜单点进去创建一个新的 Key。创建时建议给 Key 起一个能区分用途的名字比如psycopg2-debug这样后面如果要在多个项目里用方便轮换和吊销。拿到 Key 之后你需要知道 Base URL。TaoToken 的 API 端点是 https://taotoken.net/api 注意这个地址不带 UTM 参数直接用于代码里的base_url配置。Key 的格式通常是一串以sk-开头的字符串复制后先存到环境变量里不要硬编码进代码。你可以这样设置export TAOTOKEN_API_KEYsk-你的实际key如果你用的是 Windows PowerShell对应命令是$env:TAOTOKEN_API_KEYsk-你的实际key接下来是 Model ID。在控制台的模型列表里你会看到可用的模型名称比如gpt-4o、claude-3-5-sonnet等。选一个你常用的记下它的准确 ID。这个 ID 在调用时要和 Base URL、API Key 一起组成三件套。我建议你在项目根目录建一个.env文件把这三个值写进去TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的实际key TAOTOKEN_MODEL_IDgpt-4o然后用python-dotenv加载。这样做的目的是把配置和代码分离后面换 Key 或换模型不用改业务逻辑。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面有完整的请求示例和参数说明遇到 401 或模型不存在时先翻文档对照。如果你打算长期做编码类任务比如让模型持续帮你生成 SQL、审查 schema、写迁移脚本可以了解一下 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。它适合需要多轮对话、上下文保持的 Agent 场景。而如果你只是想快速验证一个模型输出用模型对话页面就够了https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。这里要强调一点TaoToken 的 Key 只用于模型调用不要把它写进 psycopg2 的数据库连接配置里。数据库连接用的是 PostgreSQL 自己的用户名密码两者完全独立。我见过有人把两个 Key 搞混结果连数据库时报认证失败排查半天才发现是复制错了。所以建议你在.env里用不同的前缀区分比如PG_和TAOTOKEN_。3. 可复制的连接池与批量写入配置骨架这一节是核心直接给可运行的代码。先看连接池。psycopg2 自带psycopg2.pool模块提供SimpleConnectionPool和ThreadedConnectionPool。前者适合单线程后者适合多线程。如果你用 Flask、FastAPI 这类多线程框架选ThreadedConnectionPool。先建一个db_config.ini把数据库连接信息放进去[postgresql] host localhost port 5432 dbname testdb user postgres password yourpassword然后写连接池管理类import configparser import psycopg2 from psycopg2 import pool from psycopg2.extras import execute_values class PgPool: _pool None classmethod def init(cls, minconn2, maxconn10): cfg configparser.ConfigParser() cfg.read(db_config.ini) section cfg[postgresql] cls._pool pool.ThreadedConnectionPool( minconnminconn, maxconnmaxconn, hostsection[host], portsection[port], dbnamesection[dbname], usersection[user], passwordsection[password] ) classmethod def get_conn(cls): return cls._pool.getconn() classmethod def put_conn(cls, conn): cls._pool.putconn(conn) classmethod def close_all(cls): cls._pool.closeall()minconn2表示池子启动时先建 2 个连接maxconn10表示最多 10 个。当并发请求超过 10 时getconn()会阻塞等待直到有连接被归还。这个数字要根据你的 PostgreSQLmax_connections和实际并发来调。一般建议maxconn不超过数据库max_connections的 70%留余量给运维连接和其他服务。接下来是批量写入。psycopg2 的executemany虽然能一次传多行但它底层还是逐条执行性能提升有限。真正快的是psycopg2.extras.execute_values它把多行拼成一条INSERT ... VALUES (...), (...), ...语句一次网络往返搞定。用法如下def batch_insert(rows, page_size1000): conn PgPool.get_conn() try: with conn.cursor() as cur: sql INSERT INTO events (user_id, event_type, payload, created_at) VALUES %s execute_values(cur, sql, rows, page_sizepage_size) conn.commit() except Exception as e: conn.rollback() raise e finally: PgPool.put_conn(conn)注意VALUES %s这里只有一个%sexecute_values会自动展开成多组值。page_size1000表示每 1000 行拼一条语句太大可能撞上 PostgreSQL 的参数上限默认 65535 个绑定参数太小则网络往返多。1000 到 5000 之间是比较稳的区间。如果你需要拿到插入后自动生成的 ID可以用RETURNING子句sql INSERT INTO events (user_id, event_type, payload) VALUES %s RETURNING id result execute_values(cur, sql, rows, fetchTrue)fetchTrue会返回所有RETURNING的结果。但要注意开了RETURNING之后page_size不宜过大因为返回结果也要占内存。事务控制方面我建议把「获取连接、执行、提交、归还」封装成一个上下文管理器避免忘记putconn导致连接泄漏from contextlib import contextmanager contextmanager def get_cursor(commitTrue): conn PgPool.get_conn() try: with conn.cursor() as cur: yield cur if commit: conn.commit() except Exception: conn.rollback() raise finally: PgPool.put_conn(conn)用的时候with get_cursor() as cur: cur.execute(SELECT count(*) FROM events) print(cur.fetchone())这样即使中间抛异常rollback和putconn都会执行。我实测下来这套骨架在 4 核 8G 的机器上单连接批量插 10 万行耗时从executemany的 40 多秒降到 3 秒左右。差距主要来自网络往返次数和事务提交次数。4. 本地验证请求与成功结果对照配置写完了怎么确认它真的在工作我习惯分三步验证连接池是否复用、批量写入是否生效、事务是否可控。第一步验证连接池复用。写一个小脚本连续获取和归还连接打印连接对象的内存地址PgPool.init(minconn2, maxconn5) for i in range(6): conn PgPool.get_conn() print(fround {i}, conn id {id(conn)}) PgPool.put_conn(conn) PgPool.close_all()如果池子工作正常你会看到前几次的id在几个值之间循环而不是每次都是新地址。这说明连接被复用了。如果每次id都不同检查是不是putconn没调用或者maxconn设得太小导致池子不断新建。第二步验证批量写入。先建一张测试表CREATE TABLE IF NOT EXISTS events ( id BIGSERIAL PRIMARY KEY, user_id INT NOT NULL, event_type TEXT NOT NULL, payload JSONB, created_at TIMESTAMPTZ DEFAULT now() );然后生成 10 万行测试数据import random rows [ (random.randint(1, 10000), click, {page: home}) for _ in range(100000) ] batch_insert(rows, page_size2000)跑完之后查一下行数with get_cursor(commitFalse) as cur: cur.execute(SELECT count(*) FROM events) print(cur.fetchone()[0])预期输出是100000。如果数字不对先看有没有异常被吞掉再检查execute_values的page_size是否超过了参数上限。第三步验证事务回滚。故意在批量插入中间抛一个异常try: with get_cursor() as cur: cur.execute(INSERT INTO events (user_id, event_type) VALUES (1, test)) raise ValueError(模拟业务异常) except ValueError: print(已回滚)然后查user_id1 AND event_typetest的记录应该是 0 条。如果查到了说明rollback没生效检查get_cursor里的异常分支。成功的结果长这样连接池日志显示连接数稳定在minconn和maxconn之间批量插入 10 万行在几秒内完成回滚后数据不残留。如果你在验证过程中需要模型帮你解释某个报错可以把错误信息贴到模型对话页面让它给出排查方向。但记住模型给的是建议最终还是要以本地实测为准。另外created_at用了TIMESTAMPTZ这是 PostgreSQL 里带时区的时间戳类型比TIMESTAMP更推荐。如果你从其他数据库迁移过来注意时区转换的差异。payload用了JSONB写入时传 JSON 字符串即可psycopg2 会自动适配。5. 本篇常见报错排查401、local proxy failed、reading choices、OAuth这一节专门对付报错。我把高频问题按错误信息分类给出原因和动作。401 Unauthorized。这个通常出现在调用模型 API 时不是数据库报错。原因有三种Key 没设置、Key 过期、Key 和 Base URL 不匹配。先检查环境变量echo $TAOTOKEN_API_KEY如果输出为空说明没加载。如果输出正常检查请求头里的Authorization是不是Bearer sk-xxx格式。Base URL 必须是https://taotoken.net/api结尾不要多加/v1或斜杠。如果还报 401去控制台重新生成一个 Key 试试。注意数据库连接报的password authentication failed是另一回事那是 PostgreSQL 的用户名密码问题别和 401 混在一起。local proxy failed。这个报错说明你的请求在本地网络层就没出去。常见原因是环境变量里配了HTTP_PROXY或HTTPS_PROXY但代理服务没启动。检查env | grep -i proxy如果有输出临时清掉unset HTTP_PROXY HTTPS_PROXY然后重试。如果你在公司内网可能需要走内部网关这时候要联系运维确认出口策略。注意这里说的是正常的网络配置问题不涉及任何绕过网络管理的手段。reading choices 相关报错。这个一般出现在解析模型返回的 JSON 时比如KeyError: choices或list index out of range。原因是返回体结构和预期不一致可能是模型返回了错误信息而不是正常结果。先打印完整响应import json resp client.chat.completions.create(...) print(json.dumps(resp.model_dump(), ensure_asciiFalse, indent2))看choices字段是否存在。如果不存在看error字段里的 message。常见的是模型 ID 写错比如把gpt-4o写成gpt4o。对照控制台的模型列表改过来。OAuth 相关报错。如果你用的是 Claude Code 或类似工具可能会遇到 OAuth token 失效。这类工具通常需要三件套Base URL、API Key、Model ID。以 Claude Code 为例配置在~/.claude/settings.json或项目级.claude/settings.json{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的实际key, ANTHROPIC_MODEL: claude-3-5-sonnet } }如果你用 Cline 或 CC Switch配置项名称可能不同但核心三件套不变。Cline 的 MCP 配置里baseUrl填https://taotoken.net/apiapiKey填你的 Keymodel填 Model ID。Codex 的auth.json里对应base_url、api_key、model三个字段。任何一处缺失或拼错都会导致认证失败。还有一个数据库侧的报错值得单独说psycopg2.OperationalError: FATAL: remaining connection slots are reserved。这是连接池maxconn设太大把 PostgreSQL 的连接槽占满了。解决办法是把maxconn调小或者调大 PostgreSQL 的max_connections。但调大max_connections会增加内存开销每个连接大约占 5 到 10 MB100 个连接就是 1 GB。所以优先从应用侧控制连接数。排查顺序建议先看报错属于网络层、认证层还是数据库层再对照上面的分类定位。不要一上来就改代码先确认环境变量和配置文件。6. 语义一致的 CTA 与后续动作代码跑通之后你可以把连接池和批量写入封装成项目里的公共模块后续所有数据库操作都走这个模块。这样做的收益是连接数可控、事务边界清晰、批量写入有统一入口。我建议在模块里加一个健康检查函数定期执行SELECT 1确认连接可用避免拿到已经断开的连接。如果你在写 SQL 或调优过程中需要快速验证语法、生成测试数据、或者让模型帮你审查索引设计可以用 TaoToken 的模型对话功能https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。它适合单次问答贴一段 SQL 进去让它解释执行计划或者让它根据你的表结构生成批量插入的测试数据。如果你需要长期做编码类任务比如持续维护数据管道、写迁移脚本、做 schema 审查Coding Plan 更合适https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。它支持多轮上下文适合 Agent 式的连续编码。API Key 的管理入口在控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。建议定期轮换 Key尤其是团队协作时每个人用独立的 Key方便审计和吊销。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 遇到参数问题先翻文档。最后给一个实用技巧把execute_values的page_size做成可配置项通过环境变量控制。不同 PostgreSQL 版本的参数上限不同线上环境调优时不用改代码改配置重启即可。另外批量写入前先ANALYZE一下目标表让查询 planner 有准确的统计信息插入性能会更稳定。这些细节看起来小但在数据量上来之后每一个都能省下可观的排查时间。
psycopg2 使用大全：从连接池到批量写入的 Python 实战配置
1. psycopg2 连接 PostgreSQL 的典型场景与高频坑psycopg2 是 Python 生态里操作 PostgreSQL 最成熟的驱动之一它能做什么简单说它把 Python 代码和 PostgreSQL 数据库之间的通信封装成一套稳定的 API让你用connect()拿连接、用cursor()执行 SQL、用commit()提交事务。适合谁适合需要做数据管道、后台服务、定时任务、报表导出、批量数据迁移的 Python 开发者。尤其是当你的数据量从几百行涨到几十万行时psycopg2 的连接池和批量写入能力就成了绕不开的话题。我见过太多项目一开始用psycopg2.connect()每次请求新建连接本地跑得好好的一上生产就报FATAL: sorry, too many clients already。PostgreSQL 默认max_connections是 100一个 Web 服务并发稍微上来连接数瞬间打满。还有人用executemany插十万行数据跑了十几分钟没结束最后发现是每条 INSERT 都单独发一次网络往返。这些问题的根子不在 SQL 写得对不对而在连接管理和批量策略没配对。这篇内容聚焦三个高频场景连接池怎么配、游标怎么用、批量写入怎么快。我会给出可直接复制的配置骨架包括psycopg2.pool和psycopg2.extras的用法再配合本地验证查询和性能对比动作。你跟着做一遍就能把「能跑」升级成「跑得稳、跑得快」。另外如果你在调试过程中需要快速验证模型生成的 SQL 或配置片段可以用 TaoToken 的模型对话功能做辅助校验地址在文末 CTA 里。先说一个容易被忽略的点psycopg2 的事务是自动开启的。你第一次调用cursor.execute()时事务就已经开始了之后所有 SQL 都在同一个事务里直到你显式commit()或rollback()。这意味着一个长时间运行的查询会持有锁阻塞其他写入。很多「数据库突然变慢」的工单最后都定位到某个忘了提交的只读连接上。所以连接池不仅要管连接数量还要管事务生命周期。还有一个坑是占位符。psycopg2 用%s作为参数占位符不是?也不是:name。你写WHERE id %s然后传 tuple驱动会做转义和类型适配。但如果你手写字符串拼接SQL 注入风险立刻上来。我建议所有值都走参数化包括LIMIT和OFFSET虽然它们看起来像整数但参数化能避免类型推断的边界问题。最后提一下autocommit。默认是False也就是手动事务模式。如果你做的是大量独立的 INSERT每条都 commit 会非常慢因为每次 commit 都是一次 fsync。这时候要么攒批要么临时开autocommit。但autocommitTrue下没有回滚保护适合日志类、监控类写入不适合有强一致要求的业务表。选哪种取决于你的数据能不能丢、能不能重放。2. TaoToken 前置准备API Key 与接入文档怎么拿在进入代码之前先把工具链准备好。TaoToken 在这里的角色是提供一个统一的模型调用入口方便你在写 psycopg2 代码时快速验证 SQL 逻辑、生成测试数据、或者让模型帮你审查连接池配置。它不是数据库也不替代 PostgreSQL而是一个辅助编码和调试的通道。第一步打开官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册并登录。登录后进入控制台地址是 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。在控制台左侧找到「API Keys」菜单点进去创建一个新的 Key。创建时建议给 Key 起一个能区分用途的名字比如psycopg2-debug这样后面如果要在多个项目里用方便轮换和吊销。拿到 Key 之后你需要知道 Base URL。TaoToken 的 API 端点是 https://taotoken.net/api 注意这个地址不带 UTM 参数直接用于代码里的base_url配置。Key 的格式通常是一串以sk-开头的字符串复制后先存到环境变量里不要硬编码进代码。你可以这样设置export TAOTOKEN_API_KEYsk-你的实际key如果你用的是 Windows PowerShell对应命令是$env:TAOTOKEN_API_KEYsk-你的实际key接下来是 Model ID。在控制台的模型列表里你会看到可用的模型名称比如gpt-4o、claude-3-5-sonnet等。选一个你常用的记下它的准确 ID。这个 ID 在调用时要和 Base URL、API Key 一起组成三件套。我建议你在项目根目录建一个.env文件把这三个值写进去TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_API_KEYsk-你的实际key TAOTOKEN_MODEL_IDgpt-4o然后用python-dotenv加载。这样做的目的是把配置和代码分离后面换 Key 或换模型不用改业务逻辑。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 里面有完整的请求示例和参数说明遇到 401 或模型不存在时先翻文档对照。如果你打算长期做编码类任务比如让模型持续帮你生成 SQL、审查 schema、写迁移脚本可以了解一下 Coding Plan入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。它适合需要多轮对话、上下文保持的 Agent 场景。而如果你只是想快速验证一个模型输出用模型对话页面就够了https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。这里要强调一点TaoToken 的 Key 只用于模型调用不要把它写进 psycopg2 的数据库连接配置里。数据库连接用的是 PostgreSQL 自己的用户名密码两者完全独立。我见过有人把两个 Key 搞混结果连数据库时报认证失败排查半天才发现是复制错了。所以建议你在.env里用不同的前缀区分比如PG_和TAOTOKEN_。3. 可复制的连接池与批量写入配置骨架这一节是核心直接给可运行的代码。先看连接池。psycopg2 自带psycopg2.pool模块提供SimpleConnectionPool和ThreadedConnectionPool。前者适合单线程后者适合多线程。如果你用 Flask、FastAPI 这类多线程框架选ThreadedConnectionPool。先建一个db_config.ini把数据库连接信息放进去[postgresql] host localhost port 5432 dbname testdb user postgres password yourpassword然后写连接池管理类import configparser import psycopg2 from psycopg2 import pool from psycopg2.extras import execute_values class PgPool: _pool None classmethod def init(cls, minconn2, maxconn10): cfg configparser.ConfigParser() cfg.read(db_config.ini) section cfg[postgresql] cls._pool pool.ThreadedConnectionPool( minconnminconn, maxconnmaxconn, hostsection[host], portsection[port], dbnamesection[dbname], usersection[user], passwordsection[password] ) classmethod def get_conn(cls): return cls._pool.getconn() classmethod def put_conn(cls, conn): cls._pool.putconn(conn) classmethod def close_all(cls): cls._pool.closeall()minconn2表示池子启动时先建 2 个连接maxconn10表示最多 10 个。当并发请求超过 10 时getconn()会阻塞等待直到有连接被归还。这个数字要根据你的 PostgreSQLmax_connections和实际并发来调。一般建议maxconn不超过数据库max_connections的 70%留余量给运维连接和其他服务。接下来是批量写入。psycopg2 的executemany虽然能一次传多行但它底层还是逐条执行性能提升有限。真正快的是psycopg2.extras.execute_values它把多行拼成一条INSERT ... VALUES (...), (...), ...语句一次网络往返搞定。用法如下def batch_insert(rows, page_size1000): conn PgPool.get_conn() try: with conn.cursor() as cur: sql INSERT INTO events (user_id, event_type, payload, created_at) VALUES %s execute_values(cur, sql, rows, page_sizepage_size) conn.commit() except Exception as e: conn.rollback() raise e finally: PgPool.put_conn(conn)注意VALUES %s这里只有一个%sexecute_values会自动展开成多组值。page_size1000表示每 1000 行拼一条语句太大可能撞上 PostgreSQL 的参数上限默认 65535 个绑定参数太小则网络往返多。1000 到 5000 之间是比较稳的区间。如果你需要拿到插入后自动生成的 ID可以用RETURNING子句sql INSERT INTO events (user_id, event_type, payload) VALUES %s RETURNING id result execute_values(cur, sql, rows, fetchTrue)fetchTrue会返回所有RETURNING的结果。但要注意开了RETURNING之后page_size不宜过大因为返回结果也要占内存。事务控制方面我建议把「获取连接、执行、提交、归还」封装成一个上下文管理器避免忘记putconn导致连接泄漏from contextlib import contextmanager contextmanager def get_cursor(commitTrue): conn PgPool.get_conn() try: with conn.cursor() as cur: yield cur if commit: conn.commit() except Exception: conn.rollback() raise finally: PgPool.put_conn(conn)用的时候with get_cursor() as cur: cur.execute(SELECT count(*) FROM events) print(cur.fetchone())这样即使中间抛异常rollback和putconn都会执行。我实测下来这套骨架在 4 核 8G 的机器上单连接批量插 10 万行耗时从executemany的 40 多秒降到 3 秒左右。差距主要来自网络往返次数和事务提交次数。4. 本地验证请求与成功结果对照配置写完了怎么确认它真的在工作我习惯分三步验证连接池是否复用、批量写入是否生效、事务是否可控。第一步验证连接池复用。写一个小脚本连续获取和归还连接打印连接对象的内存地址PgPool.init(minconn2, maxconn5) for i in range(6): conn PgPool.get_conn() print(fround {i}, conn id {id(conn)}) PgPool.put_conn(conn) PgPool.close_all()如果池子工作正常你会看到前几次的id在几个值之间循环而不是每次都是新地址。这说明连接被复用了。如果每次id都不同检查是不是putconn没调用或者maxconn设得太小导致池子不断新建。第二步验证批量写入。先建一张测试表CREATE TABLE IF NOT EXISTS events ( id BIGSERIAL PRIMARY KEY, user_id INT NOT NULL, event_type TEXT NOT NULL, payload JSONB, created_at TIMESTAMPTZ DEFAULT now() );然后生成 10 万行测试数据import random rows [ (random.randint(1, 10000), click, {page: home}) for _ in range(100000) ] batch_insert(rows, page_size2000)跑完之后查一下行数with get_cursor(commitFalse) as cur: cur.execute(SELECT count(*) FROM events) print(cur.fetchone()[0])预期输出是100000。如果数字不对先看有没有异常被吞掉再检查execute_values的page_size是否超过了参数上限。第三步验证事务回滚。故意在批量插入中间抛一个异常try: with get_cursor() as cur: cur.execute(INSERT INTO events (user_id, event_type) VALUES (1, test)) raise ValueError(模拟业务异常) except ValueError: print(已回滚)然后查user_id1 AND event_typetest的记录应该是 0 条。如果查到了说明rollback没生效检查get_cursor里的异常分支。成功的结果长这样连接池日志显示连接数稳定在minconn和maxconn之间批量插入 10 万行在几秒内完成回滚后数据不残留。如果你在验证过程中需要模型帮你解释某个报错可以把错误信息贴到模型对话页面让它给出排查方向。但记住模型给的是建议最终还是要以本地实测为准。另外created_at用了TIMESTAMPTZ这是 PostgreSQL 里带时区的时间戳类型比TIMESTAMP更推荐。如果你从其他数据库迁移过来注意时区转换的差异。payload用了JSONB写入时传 JSON 字符串即可psycopg2 会自动适配。5. 本篇常见报错排查401、local proxy failed、reading choices、OAuth这一节专门对付报错。我把高频问题按错误信息分类给出原因和动作。401 Unauthorized。这个通常出现在调用模型 API 时不是数据库报错。原因有三种Key 没设置、Key 过期、Key 和 Base URL 不匹配。先检查环境变量echo $TAOTOKEN_API_KEY如果输出为空说明没加载。如果输出正常检查请求头里的Authorization是不是Bearer sk-xxx格式。Base URL 必须是https://taotoken.net/api结尾不要多加/v1或斜杠。如果还报 401去控制台重新生成一个 Key 试试。注意数据库连接报的password authentication failed是另一回事那是 PostgreSQL 的用户名密码问题别和 401 混在一起。local proxy failed。这个报错说明你的请求在本地网络层就没出去。常见原因是环境变量里配了HTTP_PROXY或HTTPS_PROXY但代理服务没启动。检查env | grep -i proxy如果有输出临时清掉unset HTTP_PROXY HTTPS_PROXY然后重试。如果你在公司内网可能需要走内部网关这时候要联系运维确认出口策略。注意这里说的是正常的网络配置问题不涉及任何绕过网络管理的手段。reading choices 相关报错。这个一般出现在解析模型返回的 JSON 时比如KeyError: choices或list index out of range。原因是返回体结构和预期不一致可能是模型返回了错误信息而不是正常结果。先打印完整响应import json resp client.chat.completions.create(...) print(json.dumps(resp.model_dump(), ensure_asciiFalse, indent2))看choices字段是否存在。如果不存在看error字段里的 message。常见的是模型 ID 写错比如把gpt-4o写成gpt4o。对照控制台的模型列表改过来。OAuth 相关报错。如果你用的是 Claude Code 或类似工具可能会遇到 OAuth token 失效。这类工具通常需要三件套Base URL、API Key、Model ID。以 Claude Code 为例配置在~/.claude/settings.json或项目级.claude/settings.json{ env: { ANTHROPIC_BASE_URL: https://taotoken.net/api, ANTHROPIC_API_KEY: sk-你的实际key, ANTHROPIC_MODEL: claude-3-5-sonnet } }如果你用 Cline 或 CC Switch配置项名称可能不同但核心三件套不变。Cline 的 MCP 配置里baseUrl填https://taotoken.net/apiapiKey填你的 Keymodel填 Model ID。Codex 的auth.json里对应base_url、api_key、model三个字段。任何一处缺失或拼错都会导致认证失败。还有一个数据库侧的报错值得单独说psycopg2.OperationalError: FATAL: remaining connection slots are reserved。这是连接池maxconn设太大把 PostgreSQL 的连接槽占满了。解决办法是把maxconn调小或者调大 PostgreSQL 的max_connections。但调大max_connections会增加内存开销每个连接大约占 5 到 10 MB100 个连接就是 1 GB。所以优先从应用侧控制连接数。排查顺序建议先看报错属于网络层、认证层还是数据库层再对照上面的分类定位。不要一上来就改代码先确认环境变量和配置文件。6. 语义一致的 CTA 与后续动作代码跑通之后你可以把连接池和批量写入封装成项目里的公共模块后续所有数据库操作都走这个模块。这样做的收益是连接数可控、事务边界清晰、批量写入有统一入口。我建议在模块里加一个健康检查函数定期执行SELECT 1确认连接可用避免拿到已经断开的连接。如果你在写 SQL 或调优过程中需要快速验证语法、生成测试数据、或者让模型帮你审查索引设计可以用 TaoToken 的模型对话功能https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。它适合单次问答贴一段 SQL 进去让它解释执行计划或者让它根据你的表结构生成批量插入的测试数据。如果你需要长期做编码类任务比如持续维护数据管道、写迁移脚本、做 schema 审查Coding Plan 更合适https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。它支持多轮上下文适合 Agent 式的连续编码。API Key 的管理入口在控制台https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。建议定期轮换 Key尤其是团队协作时每个人用独立的 Key方便审计和吊销。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 遇到参数问题先翻文档。最后给一个实用技巧把execute_values的page_size做成可配置项通过环境变量控制。不同 PostgreSQL 版本的参数上限不同线上环境调优时不用改代码改配置重启即可。另外批量写入前先ANALYZE一下目标表让查询 planner 有准确的统计信息插入性能会更稳定。这些细节看起来小但在数据量上来之后每一个都能省下可观的排查时间。
