psycopg2 详细使用教程：从入门到精通

教程

发布日期: 2025-10-20

更新日期: 2026-11-03

文章字数: 3.6k

阅读时长: 16 分

阅读次数（新统计）: --

已读 0%

本篇目录

阅读设置

正文字号

设置仅保存在当前浏览器。

psycopg2 详细使用教程：从入门到精通

1. 什么是 psycopg2？

PostgreSQL: 一款功能极其强大的开源对象-关系型数据库系统。
psycopg2: 一个 Python 库，充当 Python 应用程序和 PostgreSQL 数据库之间的“驱动”或“适配器”。它实现了 Python DB-API 2.0 规范，允许你使用标准的 Python 代码来执行 SQL 命令。

2026-11-03 复核：Psycopg 2 当前版本为 2.9.13，仍支持旧项目；官方建议新项目优先使用当前代 Psycopg 3。本文继续保留 psycopg2 写法。

2. 安装

安装 psycopg2 有两种常见方式：

推荐 (用于开发和学习):
```
pip install psycopg2-binary
```
这个包预编译了所有必需的库，安装最简单，开箱即用。
用于生产环境 (需要编译):
```
pip install psycopg2
```
这需要你在系统上安装了 libpq（PostgreSQL 的 C 库）和相关的开发工具。两者主要是构建、依赖和维护方式不同，不能直接断言源码安装性能更好。

对于本教程，我们假设你已使用 psycopg2-binary。

3. 核心工作流：连接与游标 (Cursor)

与数据库的所有交互都遵循以下模式：

连接 (Connect): 建立到 PostgreSQL 服务器的连接。
游标 (Cursor): 创建一个游标对象。游标是执行 SQL 命令的“手柄”。
执行 (Execute): 使用游标执行 SQL 语句。
获取 (Fetch): (如果是 SELECT 语句) 从游标中取出结果。
提交 (Commit): (如果是 INSERT, UPDATE, DELETE) 提交事务，使更改永久生效。
关闭 (Close): 关闭游标和连接，释放资源。

4. 【重要】最佳实践：使用 `with` 语句

psycopg2 的两个上下文管理器职责不同：with conn: 处理事务提交或回滚，但不会关闭连接；with conn.cursor() 会关闭游标。下面用 contextlib.closing 在最外层关闭连接，内层用 with conn 管理事务。后续片段复用这里的导入和 DB_PARAMS。

这是本教程推荐的标准模式：

import psycopg2
from contextlib import closing
import sys

# 数据库连接参数
# 强烈建议使用环境变量或配置文件，不要硬编码
DB_PARAMS = {
    'host': 'localhost',      # 数据库主机
    'port': 5432,             # 端口 (默认 5432)
    'user': 'postgres',       # 你的用户名
    'password': 'your_password', # 你的密码
    'dbname': 'test_db'       # 你要连接的数据库
}

try:
    # 1. 建立连接
    # 最外层 closing 确保使用后关闭连接
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
        
            # 2. 创建游标
            # with 语句确保游标在使用后自动关闭
            with conn.cursor() as cur:
            
                # 3. 执行 SQL
                cur.execute("SELECT version();")
            
                # 4. 获取结果
                db_version = cur.fetchone() # fetchone() 获取一条结果
                print(f"数据库版本: {db_version[0]}")

                # 5. 提交 (对于 SELECT 不是必需的, 但对于 DML 是必需的)
                # conn.commit() # 在这里，with conn 会自动处理
            
except (Exception, psycopg2.DatabaseError) as error:
    print(f"连接数据库时出错: {error}")
    sys.exit(1)
    
# 6. 关闭
# cursor 的 with 关闭游标，closing 关闭连接；连接自身的 with 只管理事务
print("事务已结束，游标和连接已关闭。")

5. 【极其重要】防止 SQL 注入：参数化查询

永远不要使用 Python 的字符串格式化（如 f-string 或 %）来构建 SQL 查询。这会导致严重的安全漏洞（SQL 注入）。

错误的方式 (危险!)：

# 千万不要这样做！
user_id = "105 OR 1=1"
cur.execute(f"SELECT * FROM users WHERE id = {user_id}")

正确的方式 (安全)： psycopg2 会自动为你处理所有特殊字符的转义。你需要在 SQL 语句中使用 %s 作为占位符，然后将一个元组 (tuple) 作为第二个参数传递给 execute()。

# 正确的方式：使用占位符
user_id = "105"
query = "SELECT * FROM users WHERE id = %s"

# 单参数也需要序列，例如元组 (user_id,)；列表同样可以
cur.execute(query, (user_id,))

注意：psycopg2 统一使用 %s 作为“值”的占位符，无论数据类型是字符串、数字还是日期。
表名、列名这类 SQL 标识符不能用 %s 代替；确实需要动态生成标识符时，应使用 psycopg2.sql.Identifier 等组合方式，而不是直接拼接用户输入。

6. CRUD 操作详解

让我们创建一个 employees 表并对其进行操作。

A. 创建表 (CREATE)

create_table_query = """
CREATE TABLE IF NOT EXISTS employees (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    department VARCHAR(50),
    salary NUMERIC(10, 2),
    hire_date DATE
);
"""
try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                cur.execute(create_table_query)
                # DDL (如 CREATE TABLE) 语句也需要提交
    print("表 'employees' 已就绪，事务已提交。")
except (Exception, psycopg2.DatabaseError) as error:
    print(error)

事务提交：普通事务中的 DDL / DML 需要提交。with conn 成功退出时自动提交，异常退出时回滚；该模式下通常不需要在块内重复手动提交。

B. 插入数据 (INSERT)

插入单行

query = "INSERT INTO employees (name, department, salary, hire_date) VALUES (%s, %s, %s, %s)"
data = ("Alice Green", "Engineering", 75000.00, "2023-01-15")

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                cur.execute(query, data)
                # conn.commit() 在 with 块结束时自动调用
    print("单行插入成功。")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"插入失败: {error}")
    # 如果出错，with conn 块会引发异常，自动执行 conn.rollback()

插入多行

executemany 简化了多组参数的执行，但 psycopg2 的这个实现不保证比循环 execute 更快。大批量写入可按数据规模考虑 extras.execute_values、execute_batch 或 COPY，并结合事务边界选择。

query = "INSERT INTO employees (name, department, salary, hire_date) VALUES (%s, %s, %s, %s)"
data_list = [
    ("Bob White", "Sales", 60000.00, "2023-03-01"),
    ("Charlie Black", "Engineering", 80000.00, "2022-11-20"),
    ("David Brown", "Marketing", 55000.00, "2023-05-10")
]

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                cur.executemany(query, data_list)
                # 自动提交
    print("批量插入成功。")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"批量插入失败: {error}")

获取插入的 ID (RETURNING)

插入后经常需要获取新生成的 PRIMARY KEY。

query = "INSERT INTO employees (name, department) VALUES (%s, %s) RETURNING id"
data = ("Eva Blue", "HR")

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                cur.execute(query, data)
                new_id = cur.fetchone()[0] # 获取 RETURNING 的值
                # 正常退出事务块后自动提交
    print(f"成功插入新员工，ID: {new_id}")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"插入失败: {error}")

C. 查询数据 (SELECT)

获取单条记录 `fetchone()`

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                cur.execute("SELECT * FROM employees WHERE department = %s", ("Engineering",))
            
                print("正在获取一名工程师:")
                record = cur.fetchone() # 获取第一条
                if record:
                    print(record) # (1, 'Alice Green', 'Engineering', 75000.00, datetime.date(2023, 1, 15))

                # 再次调用 fetchone() 会获取下一条
                record2 = cur.fetchone()
                if record2:
                    print(record2)

                # 如果没有更多数据，fetchone() 返回 None
                record3 = cur.fetchone()
                print(f"第三条记录: {record3}") # None

except (Exception, psycopg2.DatabaseError) as error:
    print(error)

结果格式： 默认情况下，psycopg2 返回的数据是元组 (tuple)。

获取所有记录 `fetchall()`

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                cur.execute("SELECT name, salary FROM employees WHERE salary > %s", (58000,))
            
                print("\n薪水 > 58000 的所有员工:")
                records = cur.fetchall() # 获取所有剩余结果
            
                if records:
                    for row in records:
                        print(f"姓名: {row[0]}, 薪水: {row[1]}")
                else:
                    print("未找到记录。")
                
except (Exception, psycopg2.DatabaseError) as error:
    print(error)

注意： 如果结果集非常大，fetchall() 可能会消耗大量内存。

迭代游标 (内存高效)

普通游标仍可能在客户端缓冲整个结果集，直接 for 循环不等于流式取数。大结果集可以使用命名的服务端游标，在事务范围内分批获取：

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor(name="employees_stream") as cur:
                cur.itersize = 2000  # 命名游标每批获取的行数
                cur.execute("SELECT * FROM employees")
                print("\n--- 迭代所有员工 ---")
                for row in cur:
                    print(f"ID: {row[0]}, 姓名: {row[1]}")
                
except (Exception, psycopg2.DatabaseError) as error:
    print(error)

D. 更新数据 (UPDATE)

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                query = "UPDATE employees SET salary = salary * 1.10 WHERE department = %s"
                cur.execute(query, ("Engineering",))
            
                # cur.rowcount 包含受影响的行数
                updated_rows = cur.rowcount
                # 正常退出事务块后自动提交
    print(f"Engineering 部门 {updated_rows} 名员工已加薪。")
            
except (Exception, psycopg2.DatabaseError) as error:
    print(f"更新失败: {error}")

E. 删除数据 (DELETE)

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                query = "DELETE FROM employees WHERE name = %s"
                cur.execute(query, ("David Brown",))
            
                deleted_rows = cur.rowcount
                # 正常退出事务块后自动提交
    print(f"成功删除 {deleted_rows} 名员工。")
            
except (Exception, psycopg2.DatabaseError) as error:
    print(f"删除失败: {error}")

7. 事务管理 (Transaction)

数据库事务是一组“要么全做，要么全不做”的操作。

下面的 with conn 管理事务，外层 closing 负责关闭连接：

如果 with 块中的代码成功执行完毕（没有抛出异常），conn.commit() 会在块退出时自动调用。
如果 with 块中的代码抛出了异常，conn.rollback() 会自动调用，撤销块中已执行的所有 SQL 操作。

示例：将两条金额更新放进同一事务。 沿用 employees 表的数值列演示。UPDATE 匹配 0 行不会自动报错，因此要核对 rowcount，否则可能只更新一边。

from decimal import Decimal

def transfer_money(from_id, to_id, amount):
    try:
        amount = Decimal(str(amount))
        if not amount.is_finite() or amount <= 0:
            raise ValueError("amount 必须是正的有限数值")
        if from_id == to_id:
            raise ValueError("来源和目标不能相同")
        with closing(psycopg2.connect(**DB_PARAMS)) as conn:
            with conn:
                # conn 现在处于一个事务中
                with conn.cursor() as cur:
                
                    # 1. 从 Alice 账户扣款
                    cur.execute(
                        "UPDATE employees SET salary = salary - %s WHERE id = %s AND salary >= %s",
                        (amount, from_id, amount)
                    )
                
                    if cur.rowcount != 1:
                        raise ValueError("来源记录不存在、金额为空或余额不足")

                    # (模拟一个错误)
                    # if from_id == 1:
                    #     raise Exception("网络故障!")

                    # 2. 向 Bob 账户加款
                    cur.execute(
                        "UPDATE employees SET salary = salary + %s WHERE id = %s AND salary IS NOT NULL",
                        (amount, to_id)
                    )
                    if cur.rowcount != 1:
                        raise ValueError("目标记录不存在或金额为空")
            
        # 退出事务并关闭连接后才报告成功
        print("两条更新已提交。")
            
    except (Exception, psycopg2.DatabaseError) as error:
        # 4. 发生异常，自动 rollback
        print(f"转账失败! 错误: {error}")
        print("本次事务中的修改不提交。")

# 假设 Alice (id=1) 给 Bob (id=2) 转账 1000
transfer_money(1, 2, 1000.00)

如果你取消上面“模拟一个错误”的注释，Alice 的扣款和 Bob 的加款 都不会 发生，数据库将保持原样。异常应传播出事务块；在块内捕获后正常退出，不能再依赖异常退出触发回滚。实际应用还需自己的金额约束、并发规则和重试策略。

8. 高级技巧

A. 将结果作为字典获取 (RealDictCursor)

默认返回元组 (1, 'Alice') 不太方便。我们通常想要字典 {'id': 1, 'name': 'Alice'}。

你需要导入 psycopg2.extras 并指定 cursor_factory。

from psycopg2.extras import RealDictCursor

try:
    # 1. 在 connect 时指定 cursor_factory
    with closing(psycopg2.connect(**DB_PARAMS, cursor_factory=RealDictCursor)) as conn:
        with conn:
        
            # 2. 也可以在创建 cursor 时指定
            # with conn.cursor(cursor_factory=RealDictCursor) as cur:
        
            with conn.cursor() as cur: # cur 现在是 RealDictCursor
                cur.execute("SELECT id, name, salary FROM employees WHERE id = %s", (1,))
                record = cur.fetchone()
            
                if record:
                    print("\n--- 字典游标结果 ---")
                    print(record)
                    # 输出: {'id': 1, 'name': 'Alice Green', 'salary': 82500.00}
                    print(f"姓名: {record['name']}") # 可以通过键名访问
            
except (Exception, psycopg2.DatabaseError) as error:
    print(error)

B. 处理 `NULL` 值

psycopg2 会自动处理 Python 的 None 和 SQL 的 NULL 之间的转换。

这是你之前（在聊天记录中）遇到 invalid input syntax for "None" 错误的原因： 如果你使用 f-string (错误方式)，None 会被转成字符串 "None"。 INSERT ... VALUES ('None') -> 数据库试图将字符串 “None” 存入 double precision 字段，导致失败。

正确方式 (使用参数化): psycopg2 会将 Python 的 None 正确转换为 SQL 的 NULL。

# 插入一个 hire_date 为 NULL 的记录
query = "INSERT INTO employees (name, department, salary, hire_date) VALUES (%s, %s, %s, %s)"
data = ("Frank Nullman", "IT", 62000.00, None) # Python 的 None

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                cur.execute(query, data)
    print("成功插入带 NULL 值的记录。")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"插入 NULL 失败: {error}")

C. 处理 `JSON/JSONB` 数据

PostgreSQL 支持 JSON / JSONB。读取通常可转换为 Python 对象；写入 dict / list 时，下面显式使用 extras.Json 包装，不应认定任意字典都会自动适配为 JSON。

from psycopg2.extras import Json

# 假设表有
# ALTER TABLE employees ADD COLUMN metadata JSONB;

# 写入 JSON
meta_data = {
    'skills': ['Python', 'SQL', 'OPC-UA'],
    'manager_id': 10
}

try:
    with closing(psycopg2.connect(**DB_PARAMS)) as conn:
        with conn:
            with conn.cursor() as cur:
                # 使用 extras.Json 包装一下
                query = "UPDATE employees SET metadata = %s WHERE name = %s"
                cur.execute(query, (Json(meta_data), "Alice Green"))
            
                # 读取 JSON
                cur.execute("SELECT metadata FROM employees WHERE name = %s", ("Alice Green",))
                result = cur.fetchone()[0]
            
                print(f"\n读取到的 JSONB 数据: {result}")
                print(f"数据类型: {type(result)}") # <class 'dict'>
                print(f"技能: {result['skills'][0]}") # 'Python'
            
except (Exception, psycopg2.DatabaseError) as error:
    print(f"JSON 操作失败: {error}")

9. 总结：最佳实践清单

分清生命周期：连接的 with 管事务，游标的 with 关闭游标，closing 或 finally 负责关闭连接。
永远使用参数化查询 (cur.execute(query, (params,))) 来防止 SQL 注入。位置参数使用 %s，命名参数也可使用 %(name)s。
统一事务边界：使用 with conn 时由它提交或回滚，避免在事务中途无意手动提交。
按规模选择批量写入方式：executemany 简化调用，性能需求另看 execute_values、execute_batch 或 COPY。
使用 psycopg2.extras.RealDictCursor 来获取字典形式的结果。
正确处理 None：参数化查询会自动将 None 转为 NULL。
捕获 psycopg2.DatabaseError 来处理数据库相关的特定异常。

0xdadream

https://0xdadream.github.io/2025/10/20/psycopg2-xiang-xi-shi-yong-jiao-cheng/

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 0xdadream !

python 数据库

评论将在接近此处时加载，也可点击立即加载。

psycopg2 详细使用教程：从入门到精通

psycopg2 详细使用教程：从入门到精通

1. 什么是 psycopg2？

2. 安装

3. 核心工作流：连接与游标 (Cursor)

4. 【重要】最佳实践：使用 with 语句

5. 【极其重要】防止 SQL 注入：参数化查询

6. CRUD 操作详解

A. 创建表 (CREATE)

B. 插入数据 (INSERT)

插入单行

插入多行

获取插入的 ID (RETURNING)

C. 查询数据 (SELECT)

获取单条记录 fetchone()

获取所有记录 fetchall()

迭代游标 (内存高效)

D. 更新数据 (UPDATE)

E. 删除数据 (DELETE)

7. 事务管理 (Transaction)

8. 高级技巧

A. 将结果作为字典获取 (RealDictCursor)

B. 处理 NULL 值

C. 处理 JSON/JSONB 数据

9. 总结：最佳实践清单

你的赏识是我前进的动力

继续阅读

Python or 运算符 —— 完整深入教程

Network Camera Scanner — 完整使用与原理教程

Python 程序打包成 EXE 全面教程

4. 【重要】最佳实践：使用 `with` 语句

获取单条记录 `fetchone()`

获取所有记录 `fetchall()`

B. 处理 `NULL` 值

C. 处理 `JSON/JSONB` 数据