我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlite3.InterfaceError()。
def do_full_login(account): lock_network.acquire() time.sleep(locktime) lock_network.release() if account['type'] == 'ptc': login_ptc(account) elif account['type'] == 'google': login_google(account) new_session(account) else: lprint('[{}] Error: Login type should be either ptc or google.'.format(account['num'])) sys.exit() cursor_accs = db_accs.cursor() while True: try: cursor_accs.execute("INSERT OR REPLACE INTO accounts VALUES(?,?,?,?,?,?,?)", [account['user'], account['access_token'], account['access_expire_timestamp'], account['api_url'], 0, '0', '0']) db_accs.commit() return except sqlite3.OperationalError as e: lprint('[-] Sqlite operational error: {}, account: {} Retrying...'.format(e, account['user'])) except sqlite3.InterfaceError as e: lprint('[-] Sqlite interface error: {}, account: {} Retrying...'.format(e, account['user']))
def CheckRollbackCursorConsistency(self): """ Checks if cursors on the connection are set into a "reset" state when a rollback is done on the connection. """ con = sqlite.connect(":memory:") cur = con.cursor() cur.execute("create table test(x)") cur.execute("insert into test(x) values (5)") cur.execute("select 1 union select 2 union select 3") con.rollback() try: cur.fetchall() self.fail("InterfaceError should have been raised") except sqlite.InterfaceError, e: pass except: self.fail("InterfaceError should have been raised")
def CheckRollbackCursorConsistency(self): """ Checks if cursors on the connection are set into a "reset" state when a rollback is done on the connection. """ con = sqlite.connect(":memory:") cur = con.cursor() cur.execute("create table test(x)") cur.execute("insert into test(x) values (5)") cur.execute("select 1 union select 2 union select 3") con.rollback() try: cur.fetchall() self.fail("InterfaceError should have been raised") except sqlite.InterfaceError as e: pass except: self.fail("InterfaceError should have been raised")
def do_login(account): lprint('[{}] Login for {} account: {}'.format(account['num'], account['type'], account['user'])) timenow = get_time() cursor_accs = db_accs.cursor() db_repeat = True while db_repeat: try: acc_saved = cursor_accs.execute("SELECT access_token,access_expire_timestamp,api_url,auth_ticket__expire_timestamp_ms,auth_ticket__start,auth_ticket__end from accounts WHERE user = ?",[account['user']]).fetchone() db_repeat = False except sqlite3.OperationalError as e: lprint('[-] Sqlite operational error: {}, account: {} Retrying...'.format(e, account['user'])) except sqlite3.InterfaceError as e: lprint('[-] Sqlite interface error: {}, account: {} Retrying...'.format(e, account['user'])) if acc_saved is not None and timenow < acc_saved[1]: new_session(account) account['access_expire_timestamp'] = acc_saved[1] account['access_token'] = acc_saved[0] lprint('[{}] Reused RPC Session Token: {}'.format(account['num'], account['access_token'])) else: do_full_login(account) lprint('[{}] New RPC Session Token: {}'.format(account['num'], account['access_token'])) acc_saved = None location = LAT_C, LNG_C if acc_saved is not None and timenow < acc_saved[3]: account['api_url'] = acc_saved[2] account['auth_ticket'] = {'expire_timestamp_ms': acc_saved[3], 'start': acc_saved[4], 'end': acc_saved[5]} lprint('[{}] Reused API endpoint: {}'.format(account['num'], account['api_url'])) else: set_api_endpoint(location, account) lprint('[{}] New API endpoint: {}'.format(account['num'], account['api_url'])) if acc_tos: accept_tos(location, account) account['captcha_needed'] = False
def set_api_endpoint(location, account): account['auth_ticket'] = None get_profile(53, location, account) cursor_accs = db_accs.cursor() while True: try: cursor_accs.execute("INSERT OR REPLACE INTO accounts VALUES(?,?,?,?,?,?,?)", [account['user'], account['access_token'], account['access_expire_timestamp'], account['api_url'], account['auth_ticket']['expire_timestamp_ms'], account['auth_ticket']['start'], account['auth_ticket']['end']]) db_accs.commit() return except sqlite3.OperationalError as e: lprint('[-] Sqlite operational error: {}, account: {} Retrying...'.format(e, account['user'])) except sqlite3.InterfaceError as e: lprint('[-] Sqlite interface error: {}, account: {} Retrying...'.format(e, account['user']))
def CheckUnsupportedSeq(self): class Bar: pass val = Bar() try: self.cur.execute("insert into test(f) values (?)", (val,)) self.fail("should have raised an InterfaceError") except sqlite.InterfaceError: pass except: self.fail("should have raised an InterfaceError")
def CheckUnsupportedDict(self): class Bar: pass val = Bar() try: self.cur.execute("insert into test(f) values (:val)", {"val": val}) self.fail("should have raised an InterfaceError") except sqlite.InterfaceError: pass except: self.fail("should have raised an InterfaceError")
def CheckCursorRegistration(self): """ Verifies that subclassed cursor classes are correctly registered with the connection object, too. (fetch-across-rollback problem) """ class Connection(sqlite.Connection): def cursor(self): return Cursor(self) class Cursor(sqlite.Cursor): def __init__(self, con): sqlite.Cursor.__init__(self, con) con = Connection(":memory:") cur = con.cursor() cur.execute("create table foo(x)") cur.executemany("insert into foo(x) values (?)", [(3,), (4,), (5,)]) cur.execute("select x from foo") con.rollback() try: cur.fetchall() self.fail("should have raised InterfaceError") except sqlite.InterfaceError: pass except: self.fail("should have raised InterfaceError")
def CheckInterfaceError(self): self.assertTrue(issubclass(sqlite.InterfaceError, sqlite.Error), "InterfaceError is not a subclass of Error")
