Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _copy_exceptions method in autotest

Best Python code snippet using autotest_python

database_connection.py

Source:database_connection.py

...6_GLOBAL_CONFIG_NAMES = {7    'username' : 'user',8    'db_name' : 'database',9}10def _copy_exceptions(source, destination):11    for exception_name in _DB_EXCEPTIONS:12        setattr(destination, exception_name, getattr(source, exception_name))13class _GenericBackend(object):14    def __init__(self, database_module):15        self._database_module = database_module16        self._connection = None17        self._cursor = None18        self.rowcount = None19        _copy_exceptions(database_module, self)20    def connect(self, host=None, username=None, password=None, db_name=None):21        """22        This is assumed to enable autocommit.23        """24        raise NotImplementedError25    def disconnect(self):26        if self._connection:27            self._connection.close()28        self._connection = None29        self._cursor = None30    def execute(self, query, parameters=None):31        if parameters is None:32            parameters = ()33        self._cursor.execute(query, parameters)34        self.rowcount = self._cursor.rowcount35        return self._cursor.fetchall()36class _MySqlBackend(_GenericBackend):37    def __init__(self):38        import MySQLdb39        super(_MySqlBackend, self).__init__(MySQLdb)40    @staticmethod41    def convert_boolean(boolean, conversion_dict):42        'Convert booleans to integer strings'43        return str(int(boolean))44    def connect(self, host=None, username=None, password=None, db_name=None):45        import MySQLdb.converters46        convert_dict = MySQLdb.converters.conversions47        convert_dict.setdefault(bool, self.convert_boolean)48        self._connection = self._database_module.connect(49            host=host, user=username, passwd=password, db=db_name,50            conv=convert_dict)51        self._connection.autocommit(True)52        self._cursor = self._connection.cursor()53class _SqliteBackend(_GenericBackend):54    def __init__(self):55        from pysqlite2 import dbapi256        super(_SqliteBackend, self).__init__(dbapi2)57        self._last_insert_id_re = re.compile(r'\sLAST_INSERT_ID\(\)',58                                             re.IGNORECASE)59    def connect(self, host=None, username=None, password=None, db_name=None):60        self._connection = self._database_module.connect(db_name)61        self._connection.isolation_level = None # enable autocommit62        self._cursor = self._connection.cursor()63    def execute(self, query, parameters=None):64        # pysqlite2 uses paramstyle=qmark65        # TODO: make this more sophisticated if necessary66        query = query.replace('%s', '?')67        # pysqlite2 can't handle parameters=None (it throws a nonsense68        # exception)69        if parameters is None:70            parameters = ()71        # sqlite3 doesn't support MySQL's LAST_INSERT_ID().  Instead it has72        # something similar called LAST_INSERT_ROWID() that will do enough of73        # what we want (for our non-concurrent unittest use case).74        query = self._last_insert_id_re.sub(' LAST_INSERT_ROWID()', query)75        return super(_SqliteBackend, self).execute(query, parameters)76class _DjangoBackend(_GenericBackend):77    def __init__(self):78        from django.db import backend, connection, transaction79        super(_DjangoBackend, self).__init__(backend.Database)80        self._django_connection = connection81        self._django_transaction = transaction82    def connect(self, host=None, username=None, password=None, db_name=None):83        self._connection = self._django_connection84        self._cursor = self._connection.cursor()85    def execute(self, query, parameters=None):86        try:87            return super(_DjangoBackend, self).execute(query,88                                                       parameters=parameters)89        finally:90            self._django_transaction.commit_unless_managed()91_BACKEND_MAP = {92    'mysql': _MySqlBackend,93    'sqlite': _SqliteBackend,94    'django': _DjangoBackend,95}96class DatabaseConnection(object):97    """98    Generic wrapper for a database connection.  Supports both mysql and sqlite99    backends.100    Public attributes:101    * reconnect_enabled: if True, when an OperationalError occurs the class will102      try to reconnect to the database automatically.103    * reconnect_delay_sec: seconds to wait before reconnecting104    * max_reconnect_attempts: maximum number of time to try reconnecting before105      giving up.  Setting to RECONNECT_FOREVER removes the limit.106    * rowcount - will hold cursor.rowcount after each call to execute().107    * global_config_section - the section in which to find DB information. this108      should be passed to the constructor, not set later, and may be None, in109      which case information must be passed to connect().110    * debug - if set True, all queries will be printed before being executed111    """112    _DATABASE_ATTRIBUTES = ('db_type', 'host', 'username', 'password',113                            'db_name')114    def __init__(self, global_config_section=None, debug=False):115        self.global_config_section = global_config_section116        self._backend = None117        self.rowcount = None118        self.debug = debug119        # reconnect defaults120        self.reconnect_enabled = True121        self.reconnect_delay_sec = 20122        self.max_reconnect_attempts = 10123        self._read_options()124    def _get_option(self, name, provided_value):125        if provided_value is not None:126            return provided_value127        if self.global_config_section:128            global_config_name = _GLOBAL_CONFIG_NAMES.get(name, name)129            return global_config.global_config.get_config_value(130                self.global_config_section, global_config_name)131        return getattr(self, name, None)132    def _read_options(self, db_type=None, host=None, username=None,133                      password=None, db_name=None):134        self.db_type = self._get_option('db_type', db_type)135        self.host = self._get_option('host', host)136        self.username = self._get_option('username', username)137        self.password = self._get_option('password', password)138        self.db_name = self._get_option('db_name', db_name)139    def _get_backend(self, db_type):140        if db_type not in _BACKEND_MAP:141            raise ValueError('Invalid database type: %s, should be one of %s' %142                             (db_type, ', '.join(_BACKEND_MAP.keys())))143        backend_class = _BACKEND_MAP[db_type]144        return backend_class()145    def _reached_max_attempts(self, num_attempts):146        return (self.max_reconnect_attempts is not RECONNECT_FOREVER and147                num_attempts > self.max_reconnect_attempts)148    def _is_reconnect_enabled(self, supplied_param):149        if supplied_param is not None:150            return supplied_param151        return self.reconnect_enabled152    def _connect_backend(self, try_reconnecting=None):153        num_attempts = 0154        while True:155            try:156                self._backend.connect(host=self.host, username=self.username,157                                      password=self.password,158                                      db_name=self.db_name)159                return160            except self._backend.OperationalError:161                num_attempts += 1162                if not self._is_reconnect_enabled(try_reconnecting):163                    raise164                if self._reached_max_attempts(num_attempts):165                    raise166                traceback.print_exc()167                print ("Can't connect to database; reconnecting in %s sec" %168                       self.reconnect_delay_sec)169                time.sleep(self.reconnect_delay_sec)170                self.disconnect()171    def connect(self, db_type=None, host=None, username=None, password=None,172                db_name=None, try_reconnecting=None):173        """174        Parameters passed to this function will override defaults from global175        config.  try_reconnecting, if passed, will override176        self.reconnect_enabled.177        """178        self.disconnect()179        self._read_options(db_type, host, username, password, db_name)180        self._backend = self._get_backend(self.db_type)181        _copy_exceptions(self._backend, self)182        self._connect_backend(try_reconnecting)183    def disconnect(self):184        if self._backend:185            self._backend.disconnect()186    def execute(self, query, parameters=None, try_reconnecting=None):187        """188        Execute a query and return cursor.fetchall(). try_reconnecting, if189        passed, will override self.reconnect_enabled.190        """191        if self.debug:192            print 'Executing %s, %s' % (query, parameters)193        # _connect_backend() contains a retry loop, so don't loop here194        try:195            results = self._backend.execute(query, parameters)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.