安装 psycopg2 库

你需要安装 psycopg2，你可以使用 pip 来完成安装。

pip install psycopg2-binary

psycopg2-binary vs psycopg2

• psycopg2-binary: 这是一个预编译的版本，安装非常简单，适合大多数开发者进行开发和测试，它包含了编译好的库文件，无需在系统上安装 PostgreSQL 的开发头文件。
• psycopg2: 这是一个源码分发包，安装时需要你的系统上已经安装了 PostgreSQL 的开发头文件（如 libpq-dev 或 postgresql-devel）和 C 编译器，它通常用于生产环境或需要高度定制的场景。

对于初学者和大多数项目,使用 psycopg2-binary 是最简单直接的选择。

基本连接与查询

下面是一个最简单的示例,展示如何连接到 PostgreSQL 数据库，执行一个查询，并打印结果。

步骤 1: 准备工作

确保你的 PostgreSQL 服务正在运行，并且你已经知道以下信息：

• 主机名: 通常是 localhost（如果你在本地运行）或服务器的 IP 地址。
• 端口: 默认是 5432。
• 数据库名: 你要连接的数据库的名称。
• 用户名: 数据库的用户名。
• 密码: 对应用户的密码。

步骤 2: 编写 Python 代码

import psycopg2
# --- 1. 定义数据库连接信息 ---
# 建议在实际应用中使用环境变量或配置文件来存储这些敏感信息
DB_HOST = "localhost"
DB_NAME = "testdb"      # 你的数据库名
DB_USER = "postgres"    # 你的用户名
DB_PASS = "your_password" # 你的密码
# --- 2. 建立连接 ---
try:
    # 使用 with 语句可以自动管理连接的关闭
    conn = psycopg2.connect(
        host=DB_HOST,
        database=DB_NAME,
        user=DB_USER,
        password=DB_PASS
    )
    print("成功连接到 PostgreSQL 数据库！")
    # --- 3. 创建一个游标对象 ---
    # 游标用于执行 SQL 语句
    cur = conn.cursor()
    # --- 4. 执行一个简单的查询 ---
    cur.execute("SELECT version();")
    # --- 5. 获取查询结果 ---
    #.fetchone() 获取单行结果
    db_version = cur.fetchone()
    print(f"PostgreSQL 数据库版本: {db_version[0]}")
    # --- 6. 执行一个更复杂的查询 ---
    # 创建一个测试表 (如果不存在)
    cur.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id SERIAL PRIMARY KEY,
            name VARCHAR(50),
            email VARCHAR(50) UNIQUE
        );
    """)
    # 插入一条数据
    # 使用 %s 占位符来防止 SQL 注入，这是非常重要的安全实践！
    insert_query = "INSERT INTO users (name, email) VALUES (%s, %s);"
    user_to_insert = ('Alice', 'alice@example.com')
    cur.execute(insert_query, user_to_insert)
    # 插入多条数据
    users_to_insert = [
        ('Bob', 'bob@example.com'),
        ('Charlie', 'charlie@example.com')
    ]
    # 使用 executemany 来批量插入，效率更高
    cur.executemany(insert_query, users_to_insert)
    # 提交事务！
    # 在 psycopg2 中，默认是开启事务的，执行写操作后需要手动提交
    conn.commit()
    print("数据已成功插入到 users 表。")
    # --- 7. 查询刚刚插入的数据 ---
    cur.execute("SELECT id, name, email FROM users;")
    # fetchall() 获取所有结果
    all_users = cur.fetchall()
    print("\n当前 users 表中的所有用户:")
    for user in all_users:
        print(f"ID: {user[0]}, Name: {user[1]}, Email: {user[2]}")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"数据库操作出错: {error}")
    # 如果发生错误，回滚事务
    if conn:
        conn.rollback()
finally:
    # --- 8. 关闭游标和连接 ---
    # 确保在最后关闭游标和连接，以释放资源
    if 'cur' in locals():
        cur.close()
    if 'conn' in locals():
        conn.close()
    print("\nPostgreSQL 连接已关闭。")

使用 with 语句（推荐）

为了更安全、更方便地管理资源（连接和游标），你可以创建一个上下文管理器，这样，当 with 代码块执行完毕或发生异常时，连接和游标会自动关闭。

下面是一个重构后的示例,使用了 with 语句：

import psycopg2
from psycopg2 import sql
def get_db_connection():
    """创建并返回一个数据库连接"""
    conn = psycopg2.connect(
        host="localhost",
        database="testdb",
        user="postgres",
        password="your_password"
    )
    return conn
def query_users():
    """查询并打印所有用户"""
    try:
        # with 语句会自动管理 conn 的关闭
        with get_db_connection() as conn:
            # with 语句也会自动管理 cur 的关闭
            with conn.cursor() as cur:
                # 使用 sql 模块来安全地构建 SQL 查询，可以防止 SQL 注入
                query = sql.SQL("SELECT id, name, email FROM users ORDER BY id;")
                cur.execute(query)
                users = cur.fetchall()
                print("当前用户列表:")
                for user in users:
                    print(f"ID: {user[0]}, Name: {user[1]}, Email: {user[2]}")
    except (Exception, psycopg2.DatabaseError) as error:
        print(f"查询用户时出错: {error}")
if __name__ == "__main__":
    query_users()

使用 psycopg2.extras 获取字典形式的结果

默认情况下,fetchall() 返回的是一个元组列表，如果你希望结果以字典的形式返回，其中键是列名，可以使用 DictCursor。

import psycopg2
from psycopg2.extras import DictCursor
# ... (连接代码与之前相同) ...
try:
    with psycopg2.connect(
        host="localhost",
        database="testdb",
        user="postgres",
        password="your_password"
    ) as conn:
        # 使用 DictCursor 来获取字典形式的游标
        with conn.cursor(cursor_factory=DictCursor) as cur:
            cur.execute("SELECT id, name, email FROM users;")
            # 现在每一行都是一个字典
            users = cur.fetchall()
            print("\n当前用户列表 (字典形式):")
            for user in users:
                # 可以通过列名访问数据
                print(f"ID: {user['id']}, Name: {user['name']}, Email: {user['email']}")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"数据库操作出错: {error}")

使用连接池 (高级用法)

在高并发应用中,频繁地创建和销毁数据库连接会带来很大的性能开销，使用连接池可以复用已建立的数据库连接，显著提高性能。

psycopg2 本身不提供连接池，但你可以使用 psycopg2.pool 模块，或者更强大的第三方库如 psycopg2-pool 或 DBUtils。

使用 psycopg2.pool 的简单示例

import psycopg2
from psycopg2 import pool
# 创建一个连接池
# minconn: 最小连接数
# maxconn: 最大连接数
try:
    connection_pool = psycopg2.pool.ThreadedConnectionPool(
        minconn=1,
        maxconn=5,
        host="localhost",
        database="testdb",
        user="postgres",
        password="your_password"
    )
    print("成功创建连接池！")
    # 从连接池中获取一个连接
    conn_from_pool = connection_pool.getconn()
    print(f"从连接池获取连接: {conn_from_pool}")
    # 使用这个连接进行操作
    with conn_from_pool.cursor() as cur:
        cur.execute("SELECT %s;", ("Hello from connection pool!",))
        print(cur.fetchone()[0])
    # 将连接放回连接池
    connection_pool.putconn(conn_from_pool)
    print("连接已放回连接池。")
except (Exception, psycopg2.DatabaseError) as error:
    print(f"连接池操作出错: {error}")
finally:
    # 关闭所有连接并销毁连接池
    if 'connection_pool' in locals():
        connection_pool.closeall()
        print("所有连接已关闭，连接池已销毁。")

总结与最佳实践

1. 安装: 使用 pip install psycopg2-binary。
2. 安全: 永远不要在 SQL 查询中直接拼接字符串来传递变量，始终使用 %s 占位符和参数化查询，以防止 SQL 注入攻击。
3. 资源管理: 使用 with 语句来自动关闭游标和连接，确保资源被正确释放。
4. 事务: psycopg2 默认开启事务，执行 INSERT, UPDATE, DELETE 等写操作后，必须调用 conn.commit() 来提交更改，如果发生错误，调用 conn.rollback() 来回滚。
5. 结果格式: 对于需要按列名访问结果的场景，使用 cursor_factory=DictCursor 会非常方便。
6. 配置管理: 将数据库连接信息（如主机、用户名、密码）存储在环境变量或配置文件中，而不是硬编码在代码里。
7. 性能: 在高并发应用中，务必使用连接池来管理数据库连接。