Guaranteeing a database connection is closed after use in class?

Question

I have a wrapper class that I subclass out to the many database connections we use. I want something that will behave similarly to the try/finally functionality, but still give me the flexibility and subclassing potential of a class. Here is my base class, I would like to replace the __del__ method because I had read somewhere that there is no guarantee when this will run.

class DatabaseConnBase:
    def __init__(self, conn_func, conn_info: MutableMapping):
        self._conn = None
        self.conn_func = conn_func
        self.conn_info = conn_info

    def _initialize_connection(self):
        self._conn = self.conn_func(**self.conn_info)

    @property
    def conn(self):
        if not self._conn:
            self._initialize_connection()
        return self._conn

    @property
    def cursor(self):
        if not self._conn:
            self._initialize_connection()
        return self._conn.cursor()

    def __del__(self):
        if self._conn:
            self._conn.close()

    def commit(self):
        if not self._conn:
            raise AttributeError(
                'The connection has not been initialized, unable to commit '
                'transaction.'
            )
        self._conn.commit()

    def execute(
            self,
            query_or_stmt: str,
            verbose: bool = True,
            has_res: bool = False,
            auto_commit: bool = False
    ) -> Optional[Tuple[Any]]:
        """
        Creates a new cursor object, and executes the query/statement.  If
        `has_res` is `True`, then it returns the list of tuple results.

        :param query_or_stmt: The query or statement to run.
        :param verbose: If `True`, prints out the statement or query to STDOUT.
        :param has_res: Whether or not results should be returned.
        :param auto_commit: Immediately commits the changes to the database
         after the execute is performed.

        :return: If `has_res` is `True`, then a list of tuples.
        """
        cur = self.cursor
        if verbose:
            logger.info(f'Using {cur}')
            logger.info(f'Executing:\n{query_or_stmt}')

        cur.execute(query_or_stmt)

        if auto_commit:
            logger.info('Committing transaction...')
            self.commit()

        if has_res:
            logger.info('Returning results...')
            return cur.fetchall()

Do it explicitly using a context manager. Your class will need `__enter__` and `__exit__` methods and callers will use it via the `using` statement. — kindall, Oct 31 '17 at 20:44

score 1 · Accepted Answer · answered Oct 31 '17 at 20:44

As per the top answer on this related question: What is the __del__ method, How to call it?

The __del__ method will be called when your object is garbage collected, but as you noted, there are no guarantees on how long after destruction of references to the object this will happen.

In your case, since you check for the existence of the connection before attempting to close it, your __del__ method is safe to be called multiple times, so you could simply call it explicitly before you destroy the reference to your object.

Guaranteeing a database connection is closed after use in class?

1 Answers1