我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlite3.DatabaseError()。
def query(self, query, params=None): if self.event_db_busy: # Db not ready for queries self.log.debug("Wating for db...") self.event_db_busy.get() # Wait for event try: res = self.getDb().execute(query, params) except sqlite3.DatabaseError, err: if err.__class__.__name__ == "DatabaseError": self.log.error("Database error: %s, query: %s, try to rebuilding it..." % (err, query)) self.rebuildDb() res = self.db.cur.execute(query, params) else: raise err return res # Open file object
def ensure_exchange_schema(db): """Create all tables needed by a L{ExchangeStore}. @param db: A connection to a SQLite database. """ cursor = db.cursor() try: cursor.execute( "CREATE TABLE message_context" " (id INTEGER PRIMARY KEY, timestamp TIMESTAMP, " " secure_id TEXT NOT NULL, operation_id INTEGER NOT NULL, " " message_type text NOT NULL)") cursor.execute( "CREATE UNIQUE INDEX msgctx_operationid_idx ON " "message_context(operation_id)") except (sqlite3.OperationalError, sqlite3.DatabaseError): cursor.close() db.rollback() else: cursor.close() db.commit()
def format_cookie(self): shutil.copy2(self.cookie_file, self.copy_cookie_file) connection = sqlite3.connect(self.copy_cookie_file) cursor = connection.cursor() try: cursor.execute('SELECT host_key, path, secure, expires_utc, name, value, encrypted_value FROM cookies') except sqlite3.DatabaseError: raise Exception("Your SQLite3 package in your Python installation is out of date. " "Resolution: http://www.obsidianforensics.com/blog/upgrading-python-sqlite") with open(self.copy_cookie_txt, 'w') as file_object: file_object.write('# Netscape HTTP Cookie File\n' '# http://www.netscape.com/newsref/std/cookie_spec.html\n' '# This is a generated file! Do not edit.\n') bool_list = ['FALSE', 'TRUE'] decrypted = self.decrypt_cookie_db() for item in cursor.fetchall(): value = decrypted(item[5], item[6]) row = u'%s\t%s\t%s\t%s\t%s\t%s\t%s\n' % ( item[0], bool_list[item[0].startswith('.')], item[1], bool_list[item[2]], item[3], item[4], value) file_object.write(row) connection.close()
def retrieve(self, key, unserialize=False): retVal = None if key and (self._write_cache or os.path.isfile(self.filepath)): hash_ = HashDB.hashKey(key) retVal = self._write_cache.get(hash_) if not retVal: while True: try: for row in self.cursor.execute("SELECT value FROM storage WHERE id=?", (hash_,)): retVal = row[0] except sqlite3.OperationalError, ex: if not "locked" in ex.message: raise except sqlite3.DatabaseError, ex: errMsg = "error occurred while accessing session file '%s' ('%s'). " % (self.filepath, ex) errMsg += "If the problem persists please rerun with `--flush-session`" raise SqlmapDataException, errMsg else: break return retVal if not unserialize else unserializeObject(retVal)
def __setitem__(self, key, value): uri, metadata = value try: with sqlite3.connect(self.dbfile) as db: cursor = db.cursor() cursor.execute("PRAGMA foreign_keys=ON") dbid = cursor.execute("SELECT id FROM pyro_names WHERE name=?", (key,)).fetchone() if dbid: dbid = dbid[0] cursor.execute("DELETE FROM pyro_metadata WHERE object=?", (dbid,)) cursor.execute("DELETE FROM pyro_names WHERE id=?", (dbid,)) cursor.execute("INSERT INTO pyro_names(name, uri) VALUES(?,?)", (key, uri)) if metadata: object_id = cursor.lastrowid for m in metadata: cursor.execute("INSERT INTO pyro_metadata(object, metadata) VALUES (?,?)", (object_id, m)) cursor.close() db.commit() except sqlite3.DatabaseError as e: raise NamingError("sqlite error in setitem: " + str(e))
def optimized_metadata_search(self, metadata_all=None, metadata_any=None, return_metadata=False): try: with sqlite3.connect(self.dbfile) as db: if metadata_any: # any of the given metadata params = list(metadata_any) sql = "SELECT id, name, uri FROM pyro_names WHERE id IN (SELECT object FROM pyro_metadata WHERE metadata IN ({seq}))" \ .format(seq=",".join(['?'] * len(metadata_any))) else: # all of the given metadata params = list(metadata_all) params.append(len(metadata_all)) sql = "SELECT id, name, uri FROM pyro_names WHERE id IN (SELECT object FROM pyro_metadata WHERE metadata IN ({seq}) " \ "GROUP BY object HAVING COUNT(metadata)=?)".format(seq=",".join(['?'] * len(metadata_all))) result = db.execute(sql, params).fetchall() if return_metadata: names = {} for dbid, name, uri in result: metadata = {m[0] for m in db.execute("SELECT metadata FROM pyro_metadata WHERE object=?", (dbid,)).fetchall()} names[name] = uri, metadata else: names = {name: uri for (dbid, name, uri) in result} return names except sqlite3.DatabaseError as e: raise NamingError("sqlite error in optimized_metadata_search: " + str(e))
def import_valid_credentials(url, authentication_type, admin_panel, username, password): try: conn = sqlite3.connect(settings.SESSION_FILE) conn.execute("CREATE TABLE IF NOT EXISTS " + table_name(url) + "_creds" + \ "(id INTEGER PRIMARY KEY, url VARCHAR, authentication_type VARCHAR, admin_panel VARCHAR, "\ "username VARCHAR, password VARCHAR);") conn.execute("INSERT INTO " + table_name(url) + "_creds(url, authentication_type, "\ "admin_panel, username, password) VALUES(?,?,?,?,?)", \ (str(url), str(authentication_type), str(admin_panel), \ str(username), str(password))) conn.commit() conn.close() except sqlite3.OperationalError, err_msg: print settings.print_critical_msg(err_msg) except sqlite3.DatabaseError, err_msg: err_msg = "An error occurred while accessing session file ('" err_msg += settings.SESSION_FILE + "'). " err_msg += "If the problem persists use the '--flush-session' option." print "\n" + settings.print_critical_msg(err_msg) sys.exit(0)
def test_table_access(self): try: self.con.execute("select * from t2") except sqlite.DatabaseError, e: if not e.args[0].endswith("prohibited"): self.fail("wrong exception text: %s" % e.args[0]) return self.fail("should have raised an exception due to missing privileges")
def test_column_access(self): try: self.con.execute("select c2 from t1") except sqlite.DatabaseError, e: if not e.args[0].endswith("prohibited"): self.fail("wrong exception text: %s" % e.args[0]) return self.fail("should have raised an exception due to missing privileges")
def CheckOnConflictRollback(self): if sqlite.sqlite_version_info < (3, 2, 2): return con = sqlite.connect(":memory:") con.execute("create table foo(x, unique(x) on conflict rollback)") con.execute("insert into foo(x) values (1)") try: con.execute("insert into foo(x) values (1)") except sqlite.DatabaseError: pass con.execute("insert into foo(x) values (2)") try: con.commit() except sqlite.OperationalError: self.fail("pysqlite knew nothing about the implicit ROLLBACK")
def CheckDatabaseError(self): self.assertTrue(issubclass(sqlite.DatabaseError, sqlite.Error), "DatabaseError is not a subclass of Error")
def CheckDataError(self): self.assertTrue(issubclass(sqlite.DataError, sqlite.DatabaseError), "DataError is not a subclass of DatabaseError")
def CheckIntegrityError(self): self.assertTrue(issubclass(sqlite.IntegrityError, sqlite.DatabaseError), "IntegrityError is not a subclass of DatabaseError")
def CheckInternalError(self): self.assertTrue(issubclass(sqlite.InternalError, sqlite.DatabaseError), "InternalError is not a subclass of DatabaseError")
def CheckProgrammingError(self): self.assertTrue(issubclass(sqlite.ProgrammingError, sqlite.DatabaseError), "ProgrammingError is not a subclass of DatabaseError")
def CheckNotSupportedError(self): self.assertTrue(issubclass(sqlite.NotSupportedError, sqlite.DatabaseError), "NotSupportedError is not a subclass of DatabaseError")
def CheckExceptions(self): # Optional DB-API extension. self.assertEqual(self.cx.Warning, sqlite.Warning) self.assertEqual(self.cx.Error, sqlite.Error) self.assertEqual(self.cx.InterfaceError, sqlite.InterfaceError) self.assertEqual(self.cx.DatabaseError, sqlite.DatabaseError) self.assertEqual(self.cx.DataError, sqlite.DataError) self.assertEqual(self.cx.OperationalError, sqlite.OperationalError) self.assertEqual(self.cx.IntegrityError, sqlite.IntegrityError) self.assertEqual(self.cx.InternalError, sqlite.InternalError) self.assertEqual(self.cx.ProgrammingError, sqlite.ProgrammingError) self.assertEqual(self.cx.NotSupportedError, sqlite.NotSupportedError)
def main(db_list): """ param: db_list is a list of cadee.db files, i.e. ['cadee1.db', 'cadee2.db', ...] output: creates file OUTFILE (default: concat_cadee.db) error: will quit if OUTFILE exists """ for dbfile in db_list: # connect and get values from DB conn = sqlite3.connect(dbfile) cursor = conn.cursor() # select only 'medium' runs try: cursor.execute("SELECT * FROM results") except sqlite3.DatabaseError as e: print("Error when accessing the database: '{}' ({})".format( dbfile, e)) sys.exit(1) results = cursor.fetchall() conn.close() db = SqlDB(OUTFILE, 10000) for res in results: res = list(res) res[1] = res[1].split('_')[0] db.add_row(res) db.flush()
def CheckTableAccess(self): try: self.con.execute("select * from t2") except sqlite.DatabaseError as e: if not e.args[0].endswith("prohibited"): self.fail("wrong exception text: %s" % e.args[0]) return self.fail("should have raised an exception due to missing privileges")
def CheckColumnAccess(self): try: self.con.execute("select c2 from t1") except sqlite.DatabaseError as e: if not e.args[0].endswith("prohibited"): self.fail("wrong exception text: %s" % e.args[0]) return self.fail("should have raised an exception due to missing privileges")
