Python使用psycopg2连接PostgreSQL数据库的相关问题咨询

1. 通用数据库连接方法封装 #

可以封装，你之前遇到变量相关报错是因为没有正确管理conn、cur的生命周期，不需要用全局变量，推荐用上下文管理器封装连接逻辑，既能复用代码，还能自动处理资源释放避免泄漏。
封装示例代码如下：

import psycopg2
import sys
from psycopg2 import Error

class PostgresDB:
    def __init__(self, user, password, host, db_name):
        self.user = user
        self.password = password
        self.host = host
        self.db_name = db_name
        self.conn = None
        self.cur = None

    def __enter__(self):
        # 进入上下文时自动创建连接
        try:
            self.conn = psycopg2.connect(
                user=self.user,
                password=self.password,
                host=self.host,
                database=self.db_name
            )
            self.cur = self.conn.cursor()
            return self
        except Error as e:
            print(f"连接PostgreSQL失败: {e}")
            sys.exit(1)
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # 退出上下文时自动关闭游标、处理事务、释放连接
        if self.cur:
            self.cur.close()
        if self.conn:
            if exc_type is None:
                self.conn.commit()
            else:
                self.conn.rollback()
            self.conn.close()

    # 可按需封装通用查询、执行方法
    def query(self, sql, params=None):
        self.cur.execute(sql, params or ())
        return self.cur.fetchall()
    
    def execute(self, sql, params=None):
        self.cur.execute(sql, params or ())
        return self.cur.rowcount

调用方式非常简单，不需要重复写连接代码：

# 业务侧使用示例
with PostgresDB(userName, passwrd, hostAddr, dbName) as db:
    # 查询操作
    res = db.query("SELECT * FROM your_table WHERE id = %s", (1,))
    # 插入操作
    db.execute("INSERT INTO your_table (name) VALUES (%s)", ("test_data",))

2. 长连接方案的可行性 #

不建议直接在程序启动时创建单次连接全程复用，对于7*24小时运行的服务来说这种方案属于不良实践，存在明显缺陷：

• PostgreSQL默认配置了idle_in_transaction_session_timeout等超时参数，长时间闲置的连接会被服务端主动断开，再次调用会直接报连接失效错误
• 网络波动、数据库重启等异常都会导致连接中断，没有自动重连机制的话服务会直接不可用
• 单连接如果遇到事务未正确提交/回滚的情况，会导致后续所有操作都卡在挂起的事务中，引发死锁、数据不一致等问题

你原本的每次查询新建连接的思路不会有连接中断问题，但频繁创建销毁连接会增加数据库的性能开销，更推荐使用数据库连接池方案：
程序启动时预先创建N个可用连接放在池子里，每次操作数据库从池子里取空闲连接，用完就归还；连接池自带连接有效性检测、自动重连、连接数量管控能力，既避免了频繁建连的开销，又解决了长连接容易中断的问题。

psycopg2原生提供了连接池实现，简单示例：

from psycopg2 import pool

# 程序启动时初始化连接池
connection_pool = pool.ThreadedConnectionPool(
    minconn=2,
    maxconn=10,
    user=userName,
    password=passwrd,
    host=hostAddr,
    database=dbName
)

# 获取连接
def get_db_conn():
    conn = connection_pool.getconn()
    cur = conn.cursor()
    return conn, cur

# 用完归还连接
def return_db_conn(conn):
    if conn:
        connection_pool.putconn(conn)

# 程序退出时关闭连接池
def close_pool():
    if connection_pool:
        connection_pool.closeall()

如果确实要使用单长连接方案，必须加连接有效性校验逻辑，每次执行操作前先执行SELECT 1测试连接是否存活，失效就自动重连。

内容的提问来源于stack exchange，提问作者NicLovin