1. 零基础先跑通psycopg2 连接 PostgreSQL 到底在做什么如果你刚学完 Python 基础语法想找一个能真正落地的练手项目用 psycopg2 连接 PostgreSQL 是个很合适的选择。psycopg2 是 Python 生态里最成熟的 PostgreSQL 适配器它做的事情说白了就一件把 Python 代码里的 SQL 语句发给 PostgreSQL 数据库再把数据库返回的结果翻译成 Python 能读的列表和元组。你不需要懂数据库内核只要会写SELECT、INSERT这类基础 SQL就能用它做出一个能读能写的脚本。适合谁学三类人最合适一是刚学完 Python 想找真实项目练手的零基础学习者二是做后端开发需要跟 PostgreSQL 打交道的初级工程师三是做数据分析、需要把清洗结果写回数据库的同学。你不需要先学 SQLAlchemy 或 ORM直接上手 psycopg2 反而更能理解数据库交互的底层逻辑。我先把整条学习路线拆成四步你照着走就行第一步装好 PostgreSQL 和 psycopg2第二步用连接串连上数据库第三步用游标执行查询和写入第四步理解事务提交和回滚。这四步走完你就能独立跑通第一个数据库读写脚本。在开始之前你需要准备两样东西一个能运行的 PostgreSQL 实例本地安装或 Docker 都行以及 Python 3.8 以上的环境。如果你本地没装 PostgreSQL用 Docker 一行命令就能起一个docker run --name pg-demo -e POSTGRES_PASSWORDmysecret -e POSTGRES_DBmydb -p 5432:5432 -d postgres:16这条命令会启动一个 PostgreSQL 16 容器数据库名mydb密码mysecret端口映射到本地的 5432。等几秒钟容器起来后你就可以用 psycopg2 连上去了。这里有个零基础容易忽略的点psycopg2 本身只是客户端库它不包含数据库服务。你得先有一个正在运行的 PostgreSQL 服务psycopg2 才能连。很多人第一次报connection refused就是因为数据库服务根本没启动而不是代码写错了。另外学习过程中你可能会同时用到多个工具——比如本地跑脚本、用 AI 辅助生成 SQL、在另一个终端调试连接。这些工具如果各自维护一套凭证管理起来会很乱。我后面会讲怎么用统一的 Key/API 通道来集中管理先把数据库这条主线跑通。2. 环境准备与依赖清单psycopg2-binary 安装踩坑记录这一节把环境准备讲透因为零基础最容易卡在安装这一步。psycopg2 有两个包psycopg2和psycopg2-binary。前者需要你本地有 PostgreSQL 的开发头文件和编译工具安装时要从源码编译后者是预编译好的二进制包直接装就能用。对零基础来说无脑选psycopg2-binary省去编译的麻烦。先建一个干净的项目目录然后创建虚拟环境。虚拟环境这一步别跳过否则你系统里的包会越来越乱mkdir pg-demo cd pg-demo python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate激活虚拟环境后创建requirements.txt把依赖写进去psycopg2-binary2.9.9 python-dotenv1.0.1python-dotenv是用来读取.env文件的把数据库密码这类敏感信息从代码里分离出来这是个好习惯。然后安装pip install -r requirements.txt安装完成后验证一下是否装好python -c import psycopg2; print(psycopg2.__version__)如果输出了版本号比如2.9.9 (dt dec pq3 ext lo64)说明安装成功。如果报ModuleNotFoundError大概率是你没激活虚拟环境或者 pip 装到了别的 Python 版本下。用which python和which pip确认一下路径是否一致。接下来配置连接信息。在项目根目录创建.env文件PG_HOSTlocalhost PG_PORT5432 PG_DATABASEmydb PG_USERpostgres PG_PASSWORDmysecret注意如果你用的是 Docker 启动的 PostgreSQL默认用户是postgres密码是你启动时设的mysecret。如果你本地安装的 PostgreSQL用户和密码是你安装时设置的。这里别照抄按你自己的实际情况改。有个坑我踩过Windows 上装psycopg2-binary有时会因为缺少libpq报错。解决办法是确认你装的是 binary 版本而不是源码版本binary 版本已经把 libpq 打包进去了。如果还是报错检查一下 Python 是不是 32 位的32 位 Python 装 64 位的包会失败。环境准备好之后你还需要一个能查看数据库的工具来验证结果。可以用psql命令行也可以用 DBeaver 这类图形化工具。我习惯用psql因为快docker exec -it pg-demo psql -U postgres -d mydb进去之后执行\dt能看到当前有哪些表。现在数据库是空的下一步我们就用 Python 建表并写入数据。3. 可复制配置连接串、游标操作与参数化查询完整代码这一节是核心我给你一份可以直接复制运行的完整代码。先建一个db_demo.py把连接、建表、插入、查询、事务都串起来。先看连接部分。psycopg2 支持两种连接方式一种是传关键字参数一种是传连接串 DSN。我推荐用连接串因为配置集中、好维护import os import psycopg2 from psycopg2 import sql from dotenv import load_dotenv load_dotenv() DSN ( fhost{os.getenv(PG_HOST)} fport{os.getenv(PG_PORT)} fdbname{os.getenv(PG_DATABASE)} fuser{os.getenv(PG_USER)} fpassword{os.getenv(PG_PASSWORD)} ) def get_conn(): return psycopg2.connect(DSN)这里load_dotenv()会把.env里的变量加载到环境变量中os.getenv再读出来拼成 DSN。这样你的密码不会硬编码在代码里提交到 Git 也安全。接下来建表。用游标执行 DDL 语句def create_table(): with get_conn() as conn: with conn.cursor() as cur: cur.execute( CREATE TABLE IF NOT EXISTS users ( id SERIAL PRIMARY KEY, name VARCHAR(50) NOT NULL, email VARCHAR(100) UNIQUE NOT NULL, created_at TIMESTAMP DEFAULT NOW() ) ) print(表 users 创建完成)注意这里用了with语句管理连接和游标。with get_conn() as conn会在代码块结束时自动提交事务如果没有异常或回滚如果有异常并且关闭连接。with conn.cursor() as cur同理会自动关闭游标。这是 psycopg2 推荐的写法比手动try/except/finally简洁得多。然后是参数化查询这是重点。零基础最容易犯的错是用字符串拼接 SQL比如cur.execute(INSERT INTO users (name) VALUES ( name ))。这样写有两个问题一是 SQL 注入风险二是特殊字符比如名字里带单引号会导致语法错误。正确做法是用占位符%sdef insert_user(name, email): with get_conn() as conn: with conn.cursor() as cur: cur.execute( INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id, (name, email) ) new_id cur.fetchone()[0] print(f插入成功id{new_id}) return new_id%s是 psycopg2 的占位符不管你是字符串、数字还是日期统一用%s。psycopg2 会自动帮你做类型转换和转义安全又省心。RETURNING id是 PostgreSQL 的特性插入后直接返回自增主键省得你再查一次。查询操作def query_users(): with get_conn() as conn: with conn.cursor() as cur: cur.execute(SELECT id, name, email, created_at FROM users ORDER BY id) rows cur.fetchall() for row in rows: print(row) return rowsfetchall()返回一个列表每个元素是一个元组。如果你只想取一条用fetchone()想分批取用fetchmany(size)。数据量大的时候别用fetchall()会一次性把所有结果加载到内存用游标迭代更省内存。最后是事务控制。psycopg2 默认是自动开启事务的你执行第一条 SQL 时事务就开始了直到你调用commit()或rollback()。用with语句时正常结束自动 commit异常自动 rollback。如果你想手动控制可以这样def transfer_demo(): conn get_conn() try: with conn.cursor() as cur: cur.execute(UPDATE accounts SET balance balance - 100 WHERE id 1) cur.execute(UPDATE accounts SET balance balance 100 WHERE id 2) conn.commit() print(转账成功) except Exception as e: conn.rollback() print(f转账失败已回滚: {e}) finally: conn.close()这里两条 UPDATE 要么都成功要么都回滚保证数据一致性。这就是事务的意义。把上面的函数串起来跑if __name__ __main__: create_table() insert_user(Alice, aliceexample.com) insert_user(Bob, bobexample.com) query_users()运行python db_demo.py你应该能看到表创建成功、两条插入成功、然后打印出两行用户数据。4. 验证请求与成功结果本地跑通第一个读写脚本代码写完了怎么确认真的跑通了我给你一套验证步骤照着做就能确认每个环节都正常。第一步确认数据库服务在跑。用docker ps看容器状态或者用psql连一下docker exec -it pg-demo psql -U postgres -d mydb -c SELECT version();能输出版本信息就说明数据库正常。第二步运行脚本。在项目目录下执行python db_demo.py预期输出表 users 创建完成 插入成功id1 插入成功id2 (1, Alice, aliceexample.com, datetime.datetime(2024, 6, 1, 10, 30, 0)) (2, Bob, bobexample.com, datetime.datetime(2024, 6, 1, 10, 30, 1))如果你看到类似输出恭喜第一个读写脚本跑通了。注意created_at返回的是 Python 的datetime对象这是 psycopg2 自动做的类型映射——PostgreSQL 的timestamp映射成 Python 的datetime.datetimeinteger映射成intvarchar映射成str。你不需要手动转换。第三步去数据库里核对。用psql查一下docker exec -it pg-demo psql -U postgres -d mydb -c SELECT * FROM users;应该能看到同样的两行数据。如果 Python 里查到了但 psql 里没有那说明事务没提交——检查你的with语句是否正确包裹了execute。第四步测试参数化查询的防注入效果。故意传一个带单引号的名字insert_user(OBrien, obrienexample.com)如果用字符串拼接这里会报语法错误用%s占位符则能正常插入。这就是参数化查询的价值。第五步测试事务回滚。故意插入一个重复的 emailinsert_user(Alice2, aliceexample.com) # email 有 UNIQUE 约束会失败预期会抛出UniqueViolation异常并且因为用了with语句事务自动回滚数据库里不会多出这条记录。你可以再查一次users表确认。到这里你已经完成了连接、建表、插入、查询、事务、参数化查询的完整闭环。这套代码可以直接作为你后续项目的模板。顺便说一个实际开发中的经验当你同时用多个 AI 编码工具辅助写 SQL 或调试时每个工具可能都要配一套 API Key管理起来很烦。我后来用 TaoToken 的统一通道来集中管理这些凭证一个 Key 就能对接多个工具省得来回切换。它的 API 地址是https://taotoken.net/api你可以在控制台里生成 Key然后在各个工具里填同一个 Base URL 和 Key。这样本地脚本、AI 辅助、调试工具用的是同一套凭证不会乱。5. 本篇常见报错排查从 connection refused 到 UniqueViolation这一节把新手最常遇到的报错集中列出来每个都给你原因和解决办法。你遇到报错时先来这里对照。报错一psycopg2.OperationalError: connection to server at localhost (127.0.0.1), port 5432 failed: Connection refused这是最高频的报错。原因就一个PostgreSQL 服务没在跑或者端口不对。排查步骤先用docker ps看容器是否在运行如果容器没起用docker start pg-demo启动如果容器起了但还是连不上检查端口映射是不是 5432docker port pg-demo能看到实际映射。如果你本地装了 PostgreSQL 但没启动Windows 去服务里启动Mac 用brew services start postgresql。报错二psycopg2.OperationalError: FATAL: password authentication failed for user postgres密码错了。检查.env里的PG_PASSWORD是否和你启动容器时设的一致。如果你用的是本地安装的 PostgreSQL密码是你安装时设的。忘了密码的话Docker 容器可以删了重建本地安装的可以改pg_hba.conf把认证方式临时改成trust改完密码再改回来。报错三psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint users_email_key你插入的 email 已经存在了。这是UNIQUE约束在起作用不是 bug。解决办法要么换一个 email要么用INSERT ... ON CONFLICT (email) DO UPDATE SET name EXCLUDED.name做 upsert。这个报错也说明你的事务回滚是正常的失败的那条不会写进去。报错四psycopg2.ProgrammingError: relation users does not exist表不存在。你可能忘了先跑create_table()或者连到了错误的数据库。检查.env里的PG_DATABASE是不是你建表的那个库。用psql进去执行\dt看看当前库有哪些表。报错五ModuleNotFoundError: No module named psycopg2包没装或者装到了别的 Python 环境。先确认虚拟环境激活了命令行前面应该有(venv)前缀。然后pip list | grep psycopg2看是否装了。如果装了但还是报错用python -c import sys; print(sys.path)看 Python 的搜索路径确认虚拟环境的 site-packages 在里面。报错六psycopg2.errors.InFailedSqlTransaction: current transaction is aborted, commands ignored until end of transaction block这个报错的意思是你之前有一条 SQL 失败了事务进入了 aborted 状态后面的 SQL 全都被拒绝直到你rollback()。解决办法在except里调用conn.rollback()把事务重置。用with语句的话异常退出会自动回滚不会出现这个问题。如果你手动管理事务记得每条可能失败的 SQL 后面都要处理异常并回滚。报错七local proxy failed或401 Unauthorized如果你在用 AI 工具辅助写代码可能会遇到这类报错。401通常是 API Key 填错了或过期了去控制台重新生成一个。local proxy failed一般是本地代理配置有问题检查你的 Base URL 是否填对。如果你用的是 TaoToken 的统一通道Base URL 填https://taotoken.net/apiKey 在控制台的 API Keys 页面生成。生成后可以在模型对话页面先测一下 Key 是否可用再去配置具体工具。报错八psycopg2.ProgrammingError: argument formats cant be mixed你在execute里混用了%s和%()两种占位符。psycopg2 只认%s不管什么类型都用%s。检查你的 SQL 语句把所有占位符统一成%s。报错九psycopg2.errors.SyntaxError: syntax error at or near %你可能在 SQL 里写了LIKE %s%这种psycopg2 会把%s当成占位符解析。正确写法是用LIKE %s然后把% keyword %作为参数传进去cur.execute(SELECT * FROM users WHERE name LIKE %s, (% keyword %,))报错十psycopg2.InterfaceError: connection already closed你在连接关闭后还在用它。检查是不是在with块外面调用了cur.execute。with get_conn() as conn块结束后连接就关了后续操作要重新开连接。把这些报错对照一遍基本能覆盖你入门阶段 90% 的问题。遇到新报错先看报错类型OperationalError 是连接层ProgrammingError 是 SQL 层IntegrityError 是约束层再定位具体原因。6. 统一凭证管理与后续学习路径跑通基础脚本后你可能会想继续深入比如用连接池管理连接、用execute_values批量插入、用RealDictCursor让查询结果返回字典而不是元组。这些进阶用法我建议你按需学不用一次全啃完。先说连接池。每次psycopg2.connect()都建立一次 TCP 连接开销不小。如果你要频繁查询用psycopg2.pool或psycopg2.pool.ThreadedConnectionPool复用连接from psycopg2 import pool conn_pool pool.ThreadedConnectionPool( minconn1, maxconn10, dsnDSN ) def query_with_pool(): conn conn_pool.getconn() try: with conn.cursor() as cur: cur.execute(SELECT count(*) FROM users) return cur.fetchone()[0] finally: conn_pool.putconn(conn)getconn()从池里拿一个连接用完putconn()还回去。这样连接不会频繁创建销毁性能好很多。再说批量插入。如果你要插一万条数据逐条execute会很慢。用execute_values一次发一批from psycopg2.extras import execute_values def batch_insert(users): with get_conn() as conn: with conn.cursor() as cur: execute_values( cur, INSERT INTO users (name, email) VALUES %s, users )users是一个列表每个元素是(name, email)元组。execute_values会拼成一条多值 INSERT速度比逐条快几十倍。还有RealDictCursor让查询结果返回字典用列名访问from psycopg2.extras import RealDictCursor def query_as_dict(): with get_conn() as conn: with conn.cursor(cursor_factoryRealDictCursor) as cur: cur.execute(SELECT * FROM users) return cur.fetchall() # 返回 [{id: 1, name: Alice, ...}]这样你就不用记列的索引位置了代码可读性更好。关于学习资料我推荐三个psycopg2 官方文档的 Basic module usage 章节把连接、游标、事务讲得很清楚PostgreSQL 官方文档的 SQL Language 部分补 SQL 基础还有 psycopg2 的extras模块文档里面全是实用工具。最后说回凭证管理。你学完 psycopg2 后大概率会继续学 SQLAlchemy、FastAPI、或者用 AI 工具辅助写后端代码。这些工具各自都要配 Key管理起来很分散。我现在的做法是用 TaoToken 的统一通道一个 Key 对接多个工具。具体操作去控制台生成 Key然后在每个工具里把 Base URL 填成https://taotoken.net/apiKey 填同一个。这样你换工具时不用重新申请 Key本地脚本、AI 辅助、调试工具共用一套凭证。如果你要长期做编码和 Agent 开发可以看看 Coding Plan它把常用模型的调用额度打包了比单独配省事。你现在的任务很简单把第 3 节的代码复制下来改一下.env里的连接信息跑通它。跑通之后试着加一个DELETE操作再试着用execute_values批量插入 100 条数据。这两个练习做完psycopg2 的基础操作你就掌握了。