整个 python 应用程序中的单个数据库连接（遵循单例模式）

Question

我的问题是在整个应用程序中维护单个数据库连接的最佳方式是什么？ 使用单例模式？怎么样？

需要注意的条件：

1. 如果有多个请求，我应该使用同一个连接
2. 如果连接关闭，请创建一个新连接
3. 如果连接超时，我的代码应根据新请求创建新连接。

---

Django ORM 不支持我的数据库的驱动程序。由于同样的驱动相关问题，我使用pyodbc 连接到数据库。现在我有下面的类来创建和管理数据库连接：

class DBConnection(object):
    def __init__(self, driver, serve,
                 database, user, password):

        self.driver = driver
        self.server = server
        self.database = database
        self.user = user
        self.password = password

    def __enter__(self):
        self.dbconn = pyodbc.connect("DRIVER={};".format(self.driver) +\
                                     "SERVER={};".format(self.server) +\
                                     "DATABASE={};".format(self.database) +\
                                     "UID={};".format(self.user) +\
                                     "PWD={};".format(self.password) + \
                                     "CHARSET=UTF8",
                                     # "",
                                     ansi=True)

        return self.dbconn

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.dbconn.close()

但是这种方法的问题是它会为每个查询创建新的数据库连接。遵循单例模式的更好方法是什么？如果连接关闭，我能想到的方式将保留对连接的引用。比如：

 def get_database_connection():
     conn = DBConnection.connection
     if not conn:
          conn = DBConnection.connection = DBConnection.create_connection()
     return conn

实现这一目标的最佳方法是什么？有什么建议/想法/例子吗？

PS：我正在检查是否使用weakref，它允许创建对对象的弱引用。我认为使用weakref 和单例模式来存储连接变量是个好主意。这样，当 DB 不使用时，我就不必保持连接 alive。你们对此有何看法？

Answer 1

现在，我将继续使用单例类方法。任何看到其中潜在缺陷的人，都可以提及它们:)

DBConnector 类用于创建连接

class DBConnector(object):

   def __init__(self, driver, server, database, user, password):

        self.driver = driver
        self.server = server
        self.database = database
        self.user = user
        self.password = password
        self.dbconn = None

    # creats new connection
    def create_connection(self):
        return pyodbc.connect("DRIVER={};".format(self.driver) + \
                              "SERVER={};".format(self.server) + \
                              "DATABASE={};".format(self.database) + \
                              "UID={};".format(self.user) + \
                              "PWD={};".format(self.password) + \
                              "CHARSET=UTF8",
                              ansi=True)

    # For explicitly opening database connection
    def __enter__(self):
        self.dbconn = self.create_connection()
        return self.dbconn

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.dbconn.close()

DBConnection 类用于管理连接

class DBConnection(object):
    connection = None

    @classmethod
    def get_connection(cls, new=False):
        """Creates return new Singleton database connection"""
        if new or not cls.connection:
            cls.connection = DBConnector().create_connection()
        return cls.connection

    @classmethod
    def execute_query(cls, query):
        """execute query on singleton db connection"""
        connection = cls.get_connection()
        try:
            cursor = connection.cursor()
        except pyodbc.ProgrammingError:
            connection = cls.get_connection(new=True)  # Create new connection
            cursor = connection.cursor()
        cursor.execute(query)
        result = cursor.fetchall()
        cursor.close()
        return result

Answer 2

类 DBConnector（对象）： def __new__(cls): 如果没有 hasattr(cls, 'instance'): cls.instance = super(DBConnector, cls).__new__(cls) 返回 cls.instance def __init__(self): #构造函数中的数据库连接代码 con = DBConnector() con1 = DBConnector() con is con1 # 输出为真 希望上面的代码会有所帮助。