如何使用psycopg2在Python中连接和操作PostgreSQL数据库？

在数据驱动的应用开发中，Python与PostgreSQL的组合非常常见，而连接两者的桥梁通常就是psycopg2这个库。它实现了Python的DB-API 2.0规范，性能稳定、功能完整，是Django、SQLAlchemy等框架默认依赖的驱动之一。本文将从安装配置开始，逐步演示连接数据库、执行增删改查、处理事务以及性能优化的完整流程，帮助你快速上手并避开常见的坑。

一、安装psycopg2与建立数据库连接

psycopg2有两个发行版：psycopg2和psycopg2-binary。前者需要本地编译环境（libpq头文件、gcc等），后者是预编译版本，开箱即用。开发阶段建议直接安装二进制版：

pip install psycopg2-binary

生产环境如果对性能和稳定性有更高要求，可以安装源码版本，确保它与服务器上的PostgreSQL客户端库版本匹配。安装完成后，就可以用psycopg2.connect()建立连接，最常用的方式是通过关键字参数传参：

import psycopg2

# 建立连接
conn = psycopg2.connect(
    host="127.0.0.1",
    port=5432,
    database="testdb",
    user="postgres",
    password="yourpassword"
)

# 创建游标，用于执行SQL
cur = conn.cursor()
cur.execute("SELECT version();")
print(cur.fetchone())

# 用完及时关闭
cur.close()
conn.close()

除了关键字参数，也支持连接字符串的形式，例如psycopg2.connect("dbname=testdb user=postgres host=127.0.0.1 password=yourpassword")，两种方式效果相同，按团队习惯选择即可。连接失败时常见的报错是OperationalError，比如密码错误、服务未启动、pg_hba.conf拒绝访问等，排查时应先确认网络可达和认证配置。

二、执行SQL与参数化查询：防止注入是底线

拿到游标后，用execute()方法执行任意SQL。这里有一个必须遵守的原则：永远不要用字符串拼接构造SQL，而是使用占位符配合参数元组。psycopg2使用%s作为占位符（注意不是MySQL驱动的?）：

# 正确做法：参数化查询
cur.execute(
    "SELECT id, name, age FROM users WHERE age > %s AND name = %s",
    (18, "张三")
)

# 错误示范，严禁使用，存在SQL注入风险
# cur.execute("SELECT * FROM users WHERE name = '" + name + "'")

参数化查询不仅安全，驱动还会自动处理类型转换和特殊字符转义，比如字符串中的单引号、日期对象、None转NULL等，都不需要手动处理。对于IN子句这种参数个数不定的场景，可以利用元组自动展开的特性：

ids = [1, 3, 5, 7]
cur.execute(
    "SELECT * FROM users WHERE id IN %s",
    (tuple(ids),)  # 注意传入的是元组的元组
)

执行查询后，获取结果有三种常用方法：fetchone()取一条，fetchmany(n)取n条，fetchall()取全部。如果结果集很大，建议直接遍历游标本身，它是惰性读取的，不会一次性把所有数据加载进内存：

cur.execute("SELECT id, name FROM users")
for row in cur:
    print(f"用户ID: {row[0]}, 用户名: {row[1]}")

如果想让查询结果像字典一样按列名取值，可以传入RealDictCursor游标工厂：

from psycopg2.extras import RealDictCursor

cur = conn.cursor(cursor_factory=RealDictCursor)
cur.execute("SELECT id, name FROM users LIMIT 3")
for row in cur.fetchall():
    print(row["name"])

三、事务管理与增删改操作

psycopg2默认开启事务，执行INSERT、UPDATE、DELETE后必须调用conn.commit()才会真正落库，否则连接关闭时所有未提交的更改会被回滚。这一机制保证了原子性，但也容易让新手困惑：改了数据怎么数据库里没有？答案就是忘了提交。

try:
    cur.execute(
        "INSERT INTO users (name, age) VALUES (%s, %s)",
        ("李四", 25)
    )
    cur.execute(
        "UPDATE users SET age = %s WHERE name = %s",
        (26, "李四")
    )
    conn.commit()  # 两条语句一起提交
except Exception as e:
    conn.rollback()  # 任一步失败则整体回滚
    print(f"操作失败，已回滚: {e}")
finally:
    cur.close()
    conn.close()

更优雅的写法是把连接当上下文管理器使用，代码块正常结束自动提交，抛异常自动回滚；把游标也用with管理则自动关闭：

with conn:
    with conn.cursor() as cur:
        cur.execute("DELETE FROM users WHERE age < %s", (18,))
print("事务已自动提交")

注意一点：with conn提交或回滚事务，但并不会关闭连接，连接仍需手动close，这是文档中明确说明但常被误解的行为。

四、批量写入与性能优化技巧

逐条执行INSERT在数据量大时性能很差，每次往返都有网络开销。psycopg2提供了execute_values批量插入，内部会把多条记录合并成一条多值INSERT语句，性能提升通常在十倍以上：

from psycopg2.extras import execute_values

records = [
    ("用户A", 20),
    ("用户B", 30),
    ("用户C", 40),
]

execute_values(
    cur,
    "INSERT INTO users (name, age) VALUES %s",
    records
)
conn.commit()

如果数据量达到几十万甚至百万级，还可以配合PostgreSQL的COPY命令，把内存中的数据直接以流的形式灌入数据库，这是官方公认最快的方式：

import io
from psycopg2.extras import execute_values

buffer = io.StringIO()
for name, age in records:
    buffer.write(f"{name}\t{age}\n")
buffer.seek(0)

cur.copy_from(buffer, "users", columns=("name", "age"))
conn.commit()

另一个重要优化是连接池。Web应用频繁建立、断开数据库连接的开销不可忽视，psycopg2.pool模块提供了开箱即用的线程安全连接池：

from psycopg2.pool import ThreadedConnectionPool

pool = ThreadedConnectionPool(
    minconn=2,   # 最小连接数
    maxconn=10,  # 最大连接数
    host="127.0.0.1", database="testdb",
    user="postgres", password="yourpassword"
)

conn = pool.getconn()
try:
    with conn.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM users")
        print(cur.fetchone())
    conn.commit()
finally:
    pool.putconn(conn)  # 归还连接而不是关闭

使用连接池时切记：连接用完要putconn()归还，否则连接会泄漏，最终耗尽池子导致后续请求阻塞。

五、常见问题与最佳实践总结

实际开发中有几个高频问题值得注意。一是表名、列名不能作为参数传入占位符，%s只能用于值，动态表名需要用psycopg2.sql模块的安全拼接；二是查询中文出现乱码时，确认客户端编码与数据库编码一致，psycopg2默认使用UTF-8，一般无需额外设置；三是长事务会持有锁并阻碍VACUUM清理，业务完成后尽快提交。

from psycopg2 import sql

# 动态拼接表名（安全方式）
query = sql.SQL("SELECT * FROM {} WHERE id = %s").format(
    sql.Identifier("users")
)
cur.execute(query, (1,))

总结几条最佳实践：始终使用参数化查询杜绝SQL注入；用with管理事务和游标；大批量写入优先execute_values或COPY；常驻服务使用连接池；操作完毕确保游标和连接被正确释放。掌握这些要点后，用psycopg2操作PostgreSQL就能做到既安全又高效，为上层框架和业务逻辑打下扎实的基础。

PostgreSQLpsycopg2Python数据库操作修改时间：2026-09-07 06:14:36