[Fixed]-Single database connection throughout the python application (following singleton pattern)

7πŸ‘

For now, I am going ahead with the singleton class approach. Anyone seeing the potential flaws in this, feel to mention them πŸ™‚

DBConnector class for creating a connection

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 for managing the connections

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

3πŸ‘

class DBConnector(object):
    def __new__(cls):
        if not hasattr(cls, 'instance'):
            cls.instance = super(DBConnector, cls).__new__(cls)
        return cls.instance

    def __init__(self):
        #your db connection code in constructor

con = DBConnector()
con1 = DBConnector()
con is con1 # output is True

Hope, above code will helpful.

πŸ‘€Mahi Kumar

Leave a comment