如何使用 Python 连接 PostgreSQL 数据库？

在Python开发中，连接PostgreSQL数据库是一个常见的需求。

我们可以使用多种库来实现这一功能，其中最常用的是psycopg2。

下面我将详细介绍如何使用psycopg2来连接PostgreSQL数据库，并提供一些实际开发中的建议和注意事项。

1. 使用 `psycopg2` 连接PostgreSQL数据库

psycopg2 是一个流行的PostgreSQL数据库适配器，支持Python 2.x和3.x版本。

安装

首先，你需要安装 psycopg2 库：

pip install psycopg2

连接数据库

下面是一个简单的示例，展示如何连接到PostgreSQL数据库：

import psycopg2# 创建数据库连接
conn = psycopg2.connect(host="localhost",database="yourdatabase",user="yourusername",password="yourpassword"
)# 创建游标对象
cur = conn.cursor()# 执行SQL查询
cur.execute("SELECT * FROM yourtable")# 获取查询结果
rows = cur.fetchall()for row in rows:print(row)# 关闭游标和连接
cur.close()
conn.close()

注意事项

错误处理：在实际开发中，应该添加错误处理机制，以防止数据库连接失败或其他异常情况。

try:conn = psycopg2.connect(host="localhost",database="yourdatabase",user="yourusername",password="yourpassword")
except psycopg2.Error as e:print(f"Error: {e}")

关闭连接：在完成数据库操作后，应该关闭游标和数据库连接，以释放资源。

cur.close()
conn.close()

2. 使用上下文管理器

为了确保资源被正确释放，可以使用上下文管理器（with语句）来管理数据库连接和游标。

import psycopg2try:with psycopg2.connect(host="localhost",database="yourdatabase",user="yourusername",password="yourpassword") as conn:with conn.cursor() as cur:cur.execute("SELECT * FROM yourtable")rows = cur.fetchall()for row in rows:print(row)
except psycopg2.Error as e:print(f"Error: {e}")

3. 参数化查询

为了避免SQL注入攻击，应该使用参数化查询。

import psycopg2try:with psycopg2.connect(host="localhost",database="yourdatabase",user="yourusername",password="yourpassword") as conn:with conn.cursor() as cur:sql = "SELECT * FROM yourtable WHERE id = %s"val = (1,)cur.execute(sql, val)rows = cur.fetchall()for row in rows:print(row)
except psycopg2.Error as e:print(f"Error: {e}")

4. 使用连接池

在高并发环境下，可以使用连接池来管理数据库连接，提高性能。

from psycopg2 import pooltry:postgreSQL_pool = psycopg2.pool.SimpleConnectionPool(1,  # 最小连接数10,  # 最大连接数host="localhost",database="yourdatabase",user="yourusername",password="yourpassword")if postgreSQL_pool:print("Connection pool created")# 从连接池获取连接conn = postgreSQL_pool.getconn()if conn:print("Successfully received connection from connection pool ")cur = conn.cursor()cur.execute("SELECT * FROM yourtable")rows = cur.fetchall()for row in rows:print(row)cur.close()# 将连接放回连接池postgreSQL_pool.putconn(conn)except psycopg2.Error as e:print(f"Error: {e}")finally:if postgreSQL_pool:postgreSQL_pool.closeall()print("PostgreSQL connection pool is closed")