Python中PostgreSQL数据库连接递归创建问题咨询

问题根源分析 #

你遇到的递归创建连接错误，核心原因有三点：

1. DBConnection未实现上下文管理器协议：原类没有__enter__和__exit__方法，直接用with语句会触发非预期的实例化逻辑，导致重复创建连接
2. 模型类硬编码初始化连接：UserModel和RoleModel的构造函数都直接初始化DBConnection，当模型之间互相调用（比如UserModel中获取RoleModel数据）时，会触发连锁式的连接创建，甚至递归循环
3. 无连接复用机制：每次实例化模型都新建连接，既浪费资源，也容易引发循环依赖类的调用问题

正确写法与示例代码 #

1. 修正DBConnection类（实现上下文管理器+连接复用） #

我们将DBConnection改造为支持上下文管理器的类，同时可以结合单例模式避免重复创建连接，或者直接使用psycopg2的连接池：

方式一：上下文管理器+单例模式

import os
import psycopg2
from psycopg2.extras import RealDictCursor

class DBConnection:
    _instance = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._init_connection()
        return cls._instance

    def _init_connection(self):
        self.conn = psycopg2.connect(
            dbname=os.environ.get('DBNAME'),
            host=os.environ.get('HOST'),
            port=os.environ.get('PORT'),
            user=os.environ.get('USER'),
            password=os.environ.get('PASSWORD')
        )

    def __enter__(self):
        self.cursor = self.conn.cursor(cursor_factory=RealDictCursor)
        return self.cursor

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.conn.commit()
        else:
            self.conn.rollback()
        self.cursor.close()

    def close(self):
        if self.conn:
            self.conn.close()
            DBConnection._instance = None

方式二：使用psycopg2连接池（推荐）

连接池更适合高并发场景，能有效复用连接：

import os
import psycopg2
from psycopg2 import pool
from psycopg2.extras import RealDictCursor

class DBConnectionPool:
    _pool = None

    @classmethod
    def init_pool(cls):
        if cls._pool is None:
            cls._pool = pool.SimpleConnectionPool(
                minconn=1,
                maxconn=10,
                dbname=os.environ.get('DBNAME'),
                host=os.environ.get('HOST'),
                port=os.environ.get('PORT'),
                user=os.environ.get('USER'),
                password=os.environ.get('PASSWORD')
            )

    @classmethod
    def get_connection(cls):
        if cls._pool is None:
            cls.init_pool()
        return cls._pool.getconn()

    @classmethod
    def return_connection(cls, conn):
        cls._pool.putconn(conn)

    @classmethod
    def close_all(cls):
        if cls._pool:
            cls._pool.closeall()

# 上下文管理器封装连接池
class DBContextManager:
    def __enter__(self):
        self.conn = DBConnectionPool.get_connection()
        self.cursor = self.conn.cursor(cursor_factory=RealDictCursor)
        return self.cursor

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.conn.commit()
        else:
            self.conn.rollback()
        self.cursor.close()
        DBConnectionPool.return_connection(self.conn)

2. 修正模型类（依赖注入连接） #

不再在模型构造函数中硬编码初始化连接，而是通过参数传递连接或上下文管理器：

class UserModel:
    def __init__(self, cursor):
        self.cursor = cursor

    def update_user_data(self, user_id, data):
        query = "UPDATE users SET name = %s, email = %s WHERE id = %s"
        self.cursor.execute(query, (data['name'], data['email'], user_id))

class RoleModel:
    def __init__(self, cursor):
        self.cursor = cursor

    def get_user_role(self, user_id):
        query = "SELECT role FROM user_roles WHERE user_id = %s"
        self.cursor.execute(query, (user_id,))
        return self.cursor.fetchone()

3. 主函数调用示例 #

def main():
    # 使用上下文管理器管理连接
    with DBContextManager() as cursor:
        user_model = UserModel(cursor)
        role_model = RoleModel(cursor)

        # 获取用户角色
        role = role_model.get_user_role(1)
        print(f"User role: {role['role']}")

        # 更新用户数据
        user_model.update_user_data(1, {'name': 'Updated Name', 'email': 'updated@example.com'})

    # 程序结束时关闭连接池
    DBConnectionPool.close_all()

if __name__ == "__main__":
    main()

数据库连接最佳实践 #

• 强制使用上下文管理器：确保连接和游标自动关闭，避免资源泄漏
• 使用连接池：减少连接创建销毁的开销，提升性能，尤其适合多请求场景
• 依赖注入连接：模型类不负责创建连接，而是通过参数获取，降低耦合度，便于测试
• 避免在构造函数中初始化连接：构造函数只做初始化逻辑，连接管理交给专门的类处理
• 异常处理与事务控制：在上下文管理器中统一处理事务提交/回滚，确保数据一致性
• 单例模式（可选）：如果不需要连接池，单例模式可以避免重复创建连接实例

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