Python进行PostgreSQL数据库连接的详细使用指南

pip install psycopg2
# 或者使用二进制版本（安装更快）
pip install psycopg2-binary

# 建立连接
conn = psycopg2.connect(
    dbname="your_database",
    user="your_username",
    password="your_password",
    host="your_host",
    port="your_port"
)

# 创建游标对象
cur = conn.cursor()

# 执行简单查询
cur.execute("SELECT * FROM your_table LIMIT 5;")

# 获取结果
rows = cur.fetchall()
for row in rows:
    print(row)

# 使用参数化查询
user_id = 5
cur.execute("SELECT * FROM users WHERE id = %s;", (user_id,))
user = cur.fetchone()
print(user)

# 插入单条数据
cur.execute(
    "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id;",
    ('John Doe', 'john@example.com')
)
user_id = cur.fetchone()[0]
conn.commit()  # 必须提交事务
print(f"插入的用户ID: {user_id}")

​​​​​​​# 批量插入
users_data = [
    ('Alice', 'alice@example.com'),
    ('Bob', 'bob@example.com'),
    ('Charlie', 'charlie@example.com')
]
cur.executemany(
    "INSERT INTO users (name, email) VALUES (%s, %s);",
    users_data
)
conn.commit()

cur.execute(
    "UPDATE users SET email = %s WHERE id = %s;",
    ('new_email@example.com', 1)
)
conn.commit()

cur.execute(
    "DELETE FROM users WHERE id = %s;",
    (5,)
)
conn.commit()

# 使用with语句自动管理连接
with psycopg2.connect(
    dbname="your_database",
    user="your_username",
    password="your_password",
    host="your_host"
) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT * FROM users;")
        for row in cur:
            print(row)
    # 不需要显式调用commit()或close()，with语句会自动处理

from psycopg2 import pool

​​​​​​​# 创建连接池
connection_pool = pool.SimpleConnectionPool(
    minconn=1,
    maxconn=10,
    dbname="your_database",
    user="your_username",
    password="your_password",
    host="your_host"
)

# 从连接池获取连接
conn = connection_pool.getconn()
cur = conn.cursor()
cur.execute("SELECT * FROM users;")
# ... 执行操作 ...

​​​​​​​# 将连接返回给连接池
connection_pool.putconn(conn)

pip install sqlalchemy psycopg2-binary

from sqlalchemy import create_engine, text

# 创建引擎
engine = create_engine('postgresql://user:password@localhost:5432/dbname')

# 执行查询
with engine.connect() as connection:
    result = connection.execute(text("SELECT * FROM users;"))
    for row in result:
        print(row)