xref: /aosp_15_r20/external/autotest/frontend/db/backends/afe/base.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

from django.db.backends.mysql.base import DatabaseCreation as MySQLCreation
from django.db.backends.mysql.base import DatabaseOperations as MySQLOperations
from django.db.backends.mysql.base import DatabaseWrapper as MySQLDatabaseWrapper
from django.db.backends.mysql.base import DatabaseIntrospection as MySQLIntrospection

try:
    import MySQLdb as Database
except ImportError as e:
    from django.core.exceptions import ImproperlyConfigured
    raise ImproperlyConfigured("Error loading MySQLdb module: %s" % e)


class DatabaseOperations(MySQLOperations):
    """Custom database backend wrapper."""
    compiler_module = "autotest_lib.frontend.db.backends.afe.compiler"


class DatabaseWrapper(MySQLDatabaseWrapper):
    """Custom database backend wrapper."""

    def __init__(self, *args, **kwargs):
        self.connection = None
        super(DatabaseWrapper, self).__init__(*args, **kwargs)
        self.creation = MySQLCreation(self)
        try:
            self.ops = DatabaseOperations()
        except TypeError:
            self.ops = DatabaseOperations(connection=kwargs.get('connection'))
        self.introspection = MySQLIntrospection(self)

    def _valid_connection(self):
        if self.connection is not None:
            if self.connection.open:
                try:
                    self.connection.ping()
                    return True
                except Database.DatabaseError:
                    self.connection.close()
                    self.connection = None
        return False