PostgreSQL - Python

PostgreSQL - 使用 Psycopg 的现代 Python 接口

psycopg 是 Python 编程语言最流行的 PostgreSQL 数据库适配器。本指南重点介绍使用 Python 3 和广泛使用的 psycopg2 库的现代实践。我们还将提及它的继任者 psycopg3，它提供了进一步的改进。

安装

安装 psycopg2 的推荐方法是使用 pip（Python 的包管理器）。-binary 包包含了它自己的依赖项，是最容易上手的方式。

# 强烈建议在虚拟环境中操作
python3 -m venv venv
source venv/bin/activate

# 安装 psycopg2
pip install psycopg2-binary

注意：有一个更新的版本 psycopg3 可用，并推荐用于新项目。它可以通过 pip install psycopg 安装。尽管 API 类似，但本教程将使用 psycopg2，因为它拥有庞大的现有使用量。

连接到数据库

最佳实践是使用 with 语句来管理连接和游标。这确保即使发生错误，资源也会自动关闭。避免在代码中硬编码凭据；请使用环境变量或配置管理系统。

#!/usr/bin/env python3

import psycopg2
import os

# 最佳实践：从环境变量加载凭据
DB_NAME = os.getenv("PG_DBNAME", "testdb")
DB_USER = os.getenv("PG_USER", "postgres")
DB_PASS = os.getenv("PG_PASSWORD", "pass123")
DB_HOST = os.getenv("PG_HOST", "127.0.0.1")
DB_PORT = os.getenv("PG_PORT", "5432")

conn_string = f"dbname={DB_NAME} user={DB_USER} password={DB_PASS} host={DB_HOST} port={DB_PORT}"

try:
    # 使用 'with' 语句自动关闭连接
    with psycopg2.connect(conn_string) as conn:
        print("Opened database successfully")
        # 'with' 块在成功时还会处理 conn.commit()
        # 或在出错时处理 conn.rollback()。
except psycopg2.OperationalError as e:
    print(f"Could not connect to the database: {e}")

执行查询：CRUD 操作

重要的安全提示： 始终使用参数化查询将值传递到您的 SQL 中。切勿使用 f-字符串或字符串拼接来构建包含用户提供数据的查询，因为这会使您面临 SQL 注入攻击的风险。psycopg2 使用 %s 作为占位符，无论数据类型如何。

创建表

#!/usr/bin/env python3
import psycopg2

# (使用上一个示例中的连接字符串)

create_table_query = """
CREATE TABLE IF NOT EXISTS employees (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    age INTEGER NOT NULL,
    department TEXT,
    salary NUMERIC(10, 2)
);
"""

try:
    with psycopg2.connect(conn_string) as conn:
        with conn.cursor() as cur:
            cur.execute(create_table_query)
            print("Table 'employees' created or already exists.")
except (psycopg2.Error) as error:
    print(f"Error while connecting to PostgreSQL: {error}")

INSERT 操作（参数化）

insert_query = "INSERT INTO employees (name, age, department, salary) VALUES (%s, %s, %s, %s) RETURNING id;"
new_employee = ('Alice', 30, 'Engineering', 95000.00)

try:
    with psycopg2.connect(conn_string) as conn:
        with conn.cursor() as cur:
            cur.execute(insert_query, new_employee)
            inserted_id = cur.fetchone()[0]
            print(f"Record inserted successfully with ID: {inserted_id}")
except (psycopg2.Error) as error:
    print(f"Error during insert: {error}")

SELECT 操作

为了使代码更具可读性，您可以使用 psycopg2.extras.DictCursor 来将结果作为字典状对象而非元组获取。

from psycopg2.extras import DictCursor

select_query = "SELECT id, name, department, salary FROM employees WHERE department = %s;"

try:
    with psycopg2.connect(conn_string) as conn:
        # 使用 DictCursor 按名称访问列
        with conn.cursor(cursor_factory=DictCursor) as cur:
            cur.execute(select_query, ('Engineering',))
            rows = cur.fetchall()
            print(f"Found {cur.rowcount} employees in Engineering:")
            for row in rows:
                print(f"  ID: {row['id']}, Name: {row['name']}, Salary: {row['salary']}")
except (psycopg2.Error) as error:
    print(f"Error during select: {error}")

UPDATE 和 DELETE 操作

update_query = "UPDATE employees SET salary = %s WHERE name = %s;"
delete_query = "DELETE FROM employees WHERE name = %s;"

try:
    with psycopg2.connect(conn_string) as conn:
        with conn.cursor() as cur:
            # 更新
            cur.execute(update_query, (100000.00, 'Alice'))
            print(f"Rows updated: {cur.rowcount}")

            # 删除
            cur.execute(delete_query, ('Bob',))
            print(f"Rows deleted: {cur.rowcount}")
except (psycopg2.Error) as error:
    print(f"Error during update/delete: {error}")