### 如何使用 Python 连接 PostgreSQL 数据库 以下是通过 `psycopg2` 库实现 Python 连接到 PostgreSQL 数据库的一个完整方法及其示例代码。 #### 方法概述 `psycopg2` 是 Python 中用于连接 PostgreSQL 的适配器，提供了接近 PEP 249 标准的接口[^3]。开发者可以通过该模块执行 SQL 查询、管理事务以及处理复杂的数据类型（如 Unicode 和二进制）。以下是一个完整的连接流程： 1. 导入必要的模块。 2. 设置数据库连接参数。 3. 建立与数据库的连接。 4. 创建游标对象以执行 SQL 操作。 5. 执行查询语句并提取结果。 6. 关闭游标和连接以释放资源。 #### 示例代码 以下是一段标准的 Python 脚本，展示如何连接到 PostgreSQL 并获取数据库版本信息： ```python import psycopg2 from psycopg2 import OperationalError, Error def connect_to_database(): """ Connects to the PostgreSQL database and retrieves its version. """ connection = None try: # Connection parameters can be adjusted based on your setup connection_params = { 'dbname': 'your_db_name', 'user': 'your_username', 'password': 'your_password', 'host': 'localhost', # or IP address of the DB server 'port': '5432' # default PostgreSQL port } print("Connecting to the PostgreSQL database...") connection = psycopg2.connect(**connection_params) # Create a cursor object using the connection cursor = connection.cursor() # Execute an SQL query to get the PostgreSQL version cursor.execute('SELECT version()') # Fetch and display the result db_version = cursor.fetchone() print(f"Connected! Database Version: {db_version[0]}") except OperationalError as oe: print(f"The error '{oe}' occurred while connecting.") except Error as e: print(f"A database-related error occurred: {e}") finally: if connection is not None: connection.close() print("Database connection closed.") if __name__ == '__main__': connect_to_database() ``` 此脚本展示了基本的连接逻辑，并包含了异常处理机制以应对潜在错误情况[^1]。 #### 使用连接池优化性能 对于频繁访问数据库的应用场景，建议采用连接池技术来降低每次建立新连接的成本。以下是如何利用 `SimpleConnectionPool` 实现这一目标的例子： ```python from psycopg2 import pool try: # Initialize the connection pool with minimum and maximum connections postgresql_pool = pool.SimpleConnectionPool( minconn=1, maxconn=10, user='your_user', password='your_password', host='localhost', port='5432', database='your_db' ) if postgresql_pool: print("Connection pool created successfully!") # Get a connection from the pool conn = postgresql_pool.getconn() if conn: cursor = conn.cursor() cursor.execute("SELECT NOW()") current_time = cursor.fetchone()[0] print(f"Current time fetched from DB: {current_time}") finally: if conn: postgresql_pool.putconn(conn) # Return the connection back to the pool if postgresql_pool: postgresql_pool.closeall() # Close all pooled connections when done ``` 这种方法特别适合高并发环境下的应用开发[^4]。 --- ###