KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use set_connection_info method in lisa

Best Python code snippet using lisa_python

dbapi.py

Source:dbapi.py

...77            class_._last_server_errors[connection_key] = time.time()78            raise DBSystemException(_(u"Can't connect to database"), e)79        connection.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_READ_COMMITTED)80        connection_instance = class_._postgresql_Connection(connection, connection_data)81        class_._postgresql_reset_connection_info(connection_instance)82        connection_instance.set_connection_info('last_activity', time.time())83        cursor = connection.cursor()84        cursor.execute("show standard_conforming_strings")85        result = cursor.fetchone()[0]86        connection_instance.set_connection_info('standard_strings', (result == 'on'))87        cursor.close()88        return connection_instance89    @classmethod90    def _postgresql_close_connection(class_, connection):91        connection.connection().close()92    @classmethod93    def _postgresql_reset_connection_info(class_, connection, commands=None):94        connection.set_connection_info('transaction_commands', (commands or []))95        connection.set_connection_info('search_path', None)96        connection.set_connection_info('transaction_start_time', None)97        connection.set_connection_info('last_query_time', None)98        if __debug__:99            connection.set_connection_info('transaction_start_stack', None)100    101    def _postgresql_query(self, connection, query, outside_transaction, _retry=True):102        result = None103        def transform_arg(arg):104            if isinstance(arg, Range.Range):105                lower = arg.lower()106                upper = arg.upper()107                test_value = upper if lower is None else lower108                if isinstance(test_value, (int, long, float,)):109                    c = psycopg2.extras.NumericRange110                elif isinstance(test_value, datetime.datetime):111                    if test_value.tzinfo is None:112                        c = psycopg2.extras.DateTimeRange113                    else:114                        c = psycopg2.extras.DateTimeTZRange115                elif isinstance(test_value, datetime.date):116                    c = psycopg2.extras.DateRange117                else:118                    raise Exception("Unsupported range type", arg)119                bounds = ('[' if arg.lower_inc() else '(') + (']' if arg.upper_inc() else ')')120                arg = c(lower, upper, bounds=bounds)121            return arg122        if isinstance(query, basestring):123            query_args = {}124        else:125            query, args = query.query()126            query_args = dict([(k, transform_arg(v)) for k, v in args.items()])127        def do_query(connection, query):128            raw_connection = connection.connection()129            standard_strings = connection.connection_info('standard_strings')130            if not standard_strings:131                query = query.replace('\\', '\\\\')132            cursor = raw_connection.cursor()133            callback = self._query_callback[0]134            # The hasattr test is a hack enforced by the fact that constructor135            # calls of pytis.data classes are in very strange state now.136            if hasattr(self, '_sql_logger') and self._sql_logger is not None:137                if query_args:138                    def escape(arg):139                        if isinstance(arg, basestring):140                            result = "'%s'" % (arg.replace('\x00', '\\0').replace("'", "''"),)141                            if not standard_strings:142                                result = result.replace('\\', '\\\\')143                        elif isinstance(arg, buffer):144                            result = '<binary_data>'145                        else:146                            result = arg147                        return result148                    escaped_args = dict([(k, escape(v)) for k, v in query_args.items()])149                    query_string = query % escaped_args150                else:151                    query_string = query152                self._sql_logger.write(query_string + '\n')153            if callback is not None:154                start_time = time.time()155            # query_args shouldn't be used when empty to prevent mistaken156            # '%' processing in `query'157            try:158                if query_args:159                    cursor.execute(query, query_args)160                else:161                    cursor.execute(query)162                connection.connection_info('transaction_commands').append(query)163            finally:164                if outside_transaction:165                    raw_connection.commit()166                    self._postgresql_reset_connection_info(connection, ['commit'])167            if callback is not None:168                callback(query, start_time, time.time())169            return cursor170        def retry(message, exception):171            connection.set_connection_info('broken', True)172            if _retry:173                if not outside_transaction:174                    raise DBRetryException(message, exception, exception.args, query)175                cdata = connection.connection_data()176                search_path = connection.connection_info('search_path')177                new_connection = self._postgresql_new_connection(cdata)178                try:179                    if search_path:180                        do_query(new_connection, 'set search_path to ' + search_path)181                    result = do_query(new_connection, query)182                except Exception as e:183                    raise DBSystemException(message, e, e.args, query)184            else:185                raise DBSystemException(message, exception, exception.args, query)186            return result, new_connection187        try:188            result = do_query(connection, query)189            if query == 'commit':190                self._postgresql_commit_transaction(connection)191            elif query == 'rollback':192                self._postgresql_rollback_transaction(connection)193        except dbapi.InterfaceError as e:194            if e.args and e.args[0].find('connection already closed') != -1:195                # We believe this shouldn't happen as a program error and it196                # may occur as a result of database engine connection crash.197                log(OPERATIONAL, "Access to closed database connection")198                result, connection = retry(_(u"Database interface error"), e)199            else:200                raise DBUserException(None, e, e.args, query)201        except dbapi.NotSupportedError as e:202            if e.args and e.args[0].find('cannot perform INSERT RETURNING') != -1:203                # This is handled once again below since older dbapi versions report it as204                # ProgrammingError and newer versions as NotSupportedError.205                raise DBInsertException()206            raise DBUserException(None, e, e.args, query)207        except dbapi.ProgrammingError as e:208            if e.args:209                position = e.args[0].find210                if position('could not obtain lock') != -1:211                    raise DBLockException()212                elif position('cannot perform INSERT RETURNING') != -1:213                    raise DBInsertException()214                elif (position('error: Connection timed out') != -1 or215                      position('timeout expired') != -1):216                    217                    raise DBSystemException(_(u"Database connection timeout"), e, e.args, query)218                elif position('server closed the connection unexpectedly') != -1:219                    result, connection = retry(_(u"Database connection error"), e)220                else:221                    log(OPERATIONAL, "Transaction commands:",222                        connection.connection_info('transaction_commands'))223                    data = '%s [search_path=%s]' % (query,224                                                    connection.connection_info('search_path'),)225                    raise DBUserException(None, e, e.args, data)226            else:227                raise DBUserException(None, e, e.args, query)228        except dbapi.DataError as e:229            raise DBUserException(None, e, e.args, query)230        except dbapi.OperationalError as e:231            if e.args and e.args[0].find('could not obtain lock') != -1:232                raise DBLockException()233            result, connection = retry(_(u"Database operational error"), e)234        except dbapi.InternalError as e:235            raise DBException(None, e, query)236        except dbapi.IntegrityError as e:237            raise DBUserException(_(u"Database integrity violation"),238                                  e, e.args, query)239        now = time.time()240        connection.set_connection_info('last_query_time', now)241        connection.set_connection_info('last_activity', now)242        if connection.connection_info('transaction_start_time') is None:243            connection.set_connection_info('transaction_start_time', now)244            if __debug__:245                if query not in ('commit', 'rollback',):246                    connection.set_connection_info('transaction_start_stack',247                                                   inspect.currentframe())248        if __debug__:249            if query.startswith('fetch') or query.startswith('skip'):250                extra_info = (query,)251            else:252                self._last_informative_query = query253                extra_info = ()254            info = (time.ctime(now), self._last_informative_query,) + extra_info255            connection.set_connection_info('last_access', info)256        return self._postgresql_Result(result), connection257    def _postgresql_transform_query_result(self, result):258        cursor = result.result()259        data = []260        if cursor.description:261            for i in range(cursor.rowcount):262                row = cursor.fetchone()263                row_data = []264                for col in row:265                    if isinstance(col, psycopg2.extras.Range):266                        col = Range.Range(col.lower, col.upper,267                                          lower_inc=col.lower_inc, upper_inc=col.upper_inc)268                    elif isinstance(col, str):269                        col = unicode(col, 'utf-8')270                    row_data.append(col)271                data.append(row_data)272        else:273            data.append([cursor.rowcount])274        return PostgreSQLResult(data)275    def _postgresql_begin_transaction(self):276        # In psycopg2 `begin' is called automatically.277        # By disabling its explicit call we avoid PostgreSQL warnings about278        # transactions in progress.279        # Not only that, we even close previous transaction here.  This so that280        # contingent connection initialization commands such as coding settings281        # don't prevent to set (non-default) transaction isolation level at the282        # beginning of transaction.283        try:284            self._postgresql_commit_transaction()285        except dbapi.OperationalError as e:286            self._maybe_connection_error(e)287    288    def _postgresql_commit_transaction(self, connection=None):289        if connection is None:290            connection_, new = self._pg_get_connection()291        else:292            connection_, new = connection, False293        raw_connection = connection_.connection()294        try:295            raw_connection.commit()296            self._postgresql_reset_connection_info(connection_, ['commit'])297        except dbapi.OperationalError as e:298            self._maybe_connection_error(e)299        if new:300            self._pg_return_connection(connection_)301        302    def _postgresql_rollback_transaction(self, connection=None):303        if connection is None:304            connection_, new = self._pg_get_connection()305        else:306            connection_, new = connection, False307        raw_connection = connection_.connection()308        try:309            raw_connection.rollback()310        except dbapi.OperationalError as e:311            self._maybe_connection_error(e)312        # For unknown reasons, connection client encoding gets reset after313        # rollback314        self._postgresql_reset_connection_info(connection_, ['rollback'])315        cursor = raw_connection.cursor()316        query = 'set client_encoding to "utf-8"'317        try:318            cursor.execute(query)319        except dbapi.OperationalError as e:320            self._maybe_connection_error(e)321        # We commit the transaction immediately to prevent an open322        # (maybe) idle transaction.323        raw_connection.commit()324        if new:325            self._pg_return_connection(connection_)326    def _maybe_connection_error(self, e):327        if e.args[0].find('server closed the connection unexpectedly') != -1:328            raise DBSystemException(_(u"Database connection error"), e, e.args)...

apireg.py

Source:apireg.py

...50    # Nothing in user's registry, try system registry51    if user and not url:52        url = get_connection_info(user=False)53    return url54def set_connection_info(url, user=True):55    """ Set API info in register """56    section, keypath = reg_info(user)57    try:58        hive = _winreg.ConnectRegistry(None, section)59        try:60            key = _winreg.CreateKey(hive, keypath)61        except:62            pass63        key = _winreg.OpenKey(hive, keypath)64        mykey = _winreg.CreateKey(key, 'api')65        _winreg.SetValueEx(mykey, 'url', None, _winreg.REG_SZ, url)66        _winreg.CloseKey(mykey)67        _winreg.CloseKey(key)68    except WindowsError:69        if user:70            set_connection_info(url, user=False)71        pass72    finally:73        _winreg.CloseKey(hive)74def del_connection_info(user=True):75    """ Remove API info from register """76    section, keypath = reg_info(user)77    try:78        hive = _winreg.ConnectRegistry(None, section)79        key = _winreg.OpenKey(hive, keypath)80        _winreg.DeleteKey(key, 'api')81        _winreg.CloseKey(key)82    except WindowsError:83        if user:84            del_connection_info(user=False)85        pass86    finally:87        _winreg.CloseKey(hive)88def get_install_lng():89    """ Return language-code used by the installer """90    lng = 091    try:92        hive = _winreg.ConnectRegistry(None, _winreg.HKEY_LOCAL_MACHINE)93        key = _winreg.OpenKey(hive, r"Software\SABnzbd")94        for i in range(0, _winreg.QueryInfoKey(key)[1]):95            name, value, val_type = _winreg.EnumValue(key, i)96            if name == 'Installer Language':97                lng = value98        _winreg.CloseKey(key)99    except WindowsError:100        pass101    finally:102        _winreg.CloseKey(hive)103    return lng104if __name__ == '__main__':105    print 'URL = %s' % get_connection_info()106    print 'Language = %s' % get_install_lng()107    # del_connection_info()...

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.