我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用sqlite3.Cursor()。
def CheckCursorConstructorCallCheck(self): """ Verifies that cursor methods check whether base class __init__ was called. """ class Cursor(sqlite.Cursor): def __init__(self, con): pass con = sqlite.connect(":memory:") cur = Cursor(con) try: cur.execute("select 4+5").fetchall() self.fail("should have raised ProgrammingError") except sqlite.ProgrammingError: pass except: self.fail("should have raised ProgrammingError")
def CheckCursorConstructorCallCheck(self): """ Verifies that cursor methods check wether base class __init__ was called. """ class Cursor(sqlite.Cursor): def __init__(self, con): pass con = sqlite.connect(":memory:") cur = Cursor(con) try: cur.execute("select 4+5").fetchall() self.fail("should have raised ProgrammingError") except sqlite.ProgrammingError: pass except: self.fail("should have raised ProgrammingError")
def execute_insertion(self, query, *args): """Executes a query and returns the entire cursor This is intended to return the cursor, which needs to be closed after the results are fetched. This method can be used both for INSERT SQL statement or others like SELECT. :param string query: The SQL query. :return: A cursor that must be closed :rtype: sqlite3.Cursor """ connection = self.connection # The row factory returns a richer object to user, similar to dict connection.row_factory = sqlite3.Row cursor = connection.cursor() # Before execute a SQL query is necessary to turn on # the foreign_keys restrictions # cursor.execute("PRAGMA foreign_keys = ON;") # Execute the *real* query cursor.execute(query, args) connection.commit() return cursor # Must be closed outside function
def __call__(self, func): def wrapper(*args, **kwargs): index = 0 args = list(args) for arg in args[:2]: if arg == DB or isinstance(arg, DB): index = 1 elif isinstance(arg, sqlite3.Cursor): if arg.connection == DB.connection: return func(*args, **kwargs) else: args.pop(index) break else: break if DB.connection: args.insert(index, DB.cursor) return func(*args, **kwargs) else: with self as cur: args.insert(index, cur) return func(*args, **kwargs) return wrapper
def insert_article(curs: sqlite3.Cursor, article: newspaper.Article, table: str) -> None: """Insert relevant article fields into db table.""" if table == 'articles' or table == 'ground_truth': authors = ','.join(article.authors) try: publish_date = article.publish_date.isoformat() except AttributeError: publish_date = '' tags = ','.join(article.tags) fields = (article.title, authors, publish_date, article.url, article.text, tags) if table == 'articles': command = 'INSERT INTO articles VALUES (?, ?, ?, ?, ?, ?)' else: fields += (article.label, ) command = 'INSERT INTO ground_truth VALUES (?, ?, ?, ?, ?, ?, ?)' curs.execute(command, fields) elif table == 'bad_articles': curs.execute('INSERT INTO bad_articles VALUES (?)', (article.url, )) elif table == 'old_articles': curs.execute('INSERT INTO old_articles VALUES (?)', (article.url, ))
def _DedupingEntityGenerator(cursor): """Generator that removes duplicate entities from the results. Generate datastore entities from a cursor, skipping the duplicates Args: cursor: a SQLite3.Cursor or subclass. Yields: Entities that do not share a key. """ seen = set() for row in cursor: row_key, row_entity = row[:2] encoded_row_key = str(row_key) if encoded_row_key in seen: continue seen.add(encoded_row_key) storage_entity = entity_pb.EntityProto(row_entity) record = datastore_stub_util._FromStorageEntity(storage_entity) record = datastore_stub_util.LoadRecord(record) yield record
def select(self, select, table_name, where=None, extra=None): """ Send a SELECT query to the database. :param str select: Attribute for the ``SELECT`` query. :param str table_name: |arg_select_table_name| :param str where: |arg_select_where| :param str extra: |arg_select_extra| :return: Result of the query execution. :rtype: sqlite3.Cursor :raises simplesqlite.NullDatabaseConnectionError: |raises_check_connection| :raises simplesqlite.TableNotFoundError: |raises_verify_table_existence| :raises simplesqlite.OperationalError: |raises_operational_error| .. seealso:: :py:meth:`.sqlquery.SqlQuery.make_select` """ self.verify_table_existence(table_name) query = SqlQuery.make_select(select, table_name, where, extra) return self.execute_query(query, logging.getLogger().findCaller())
def constructs_correctly(self): self.assertIsInstance(self.logdb.conn, sqlite3.Connection) self.assertIsInstance(self.logdb.cursor, sqlite3.Cursor)
def __init__(self, *args, **kwargs): sqlite.Cursor.__init__(self, *args, **kwargs) self.row_factory = dict_factory
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 CheckCursorWrongClass(self): class Foo: pass foo = Foo() try: cur = sqlite.Cursor(foo) self.fail("should have raised a ValueError") except TypeError: pass
def CheckFakeCursorClass(self): # Issue #24257: Incorrect use of PyObject_IsInstance() caused # segmentation fault. class FakeCursor(str): __class__ = sqlite.Cursor cur = self.con.cursor(factory=FakeCursor) self.assertRaises(TypeError, sqlite.Row, cur, ())
def _get_tuple_generator(cursor: sqlite3.Cursor, fetch_size=100): while True: tuples = cursor.fetchmany(fetch_size) if not tuples: break yield from tuples
def __init__(self, cursor: sqlite3.Cursor): self.cursor = cursor self._keys = None
def CheckIsInstance(self): cur = self.con.cursor() self.assertIsInstance(cur, sqlite.Cursor) cur = self.con.cursor(MyCursor) self.assertIsInstance(cur, MyCursor) cur = self.con.cursor(factory=lambda con: MyCursor(con)) self.assertIsInstance(cur, MyCursor)
def CheckFakeCursorClass(self): # Issue #24257: Incorrect use of PyObject_IsInstance() caused # segmentation fault. # Issue #27861: Also applies for cursor factory. class FakeCursor(str): __class__ = sqlite.Cursor self.con.row_factory = sqlite.Row self.assertRaises(TypeError, self.con.cursor, FakeCursor) self.assertRaises(TypeError, sqlite.Row, FakeCursor(), ())
def addfield(self, fieldname, fieldtype='text', initialvalue=None): if not fieldtype in fieldtype_enum: raise ValueError("unsupported field type: "+fieldtype) self._con.execute("alter table skirtruns add column " + fieldname + " " + fieldtype) if initialvalue!=None: self._con.execute("update skirtruns set " + fieldname + " = ?", (initialvalue,)) self._con.commit() ## This function executes an arbitrary SQL statement on the SKIRT-runs database and returns an sqlite3.Cursor # object representing the result. It is \em not intended for use in production code.
def cursor(self, expr = None, *args) -> sqlite3.Cursor: if verbose(3): pargs = [str(arg) for arg in args] stack = [f"{os.path.basename(frame.filename)}:{frame.lineno} {frame.name}" for frame in traceback.extract_stack(limit = 15)] width = max(len(expr or ''), *[len(i) for i in pargs], *[len(i) for i in stack]) print(file = sys.stderr) print(u'\u250c' + u'\u2500' * (width + 2) + u'\u2510', file = sys.stderr) print(u"\u2502 %*s \u2502" % (-width, expr or ''), file = sys.stderr) if pargs: print(u'\u251c' + u'\u2500' * (width + 2) + u'\u2524', file = sys.stderr) for arg in pargs: print(u"\u2502 %*s \u2502" % (-width, arg), file = sys.stderr) print(u'\u251c' + u'\u2500' * (width + 2) + u'\u2524', file = sys.stderr) else: print(u'\u255e' + u'\u2550' * (width + 2) + u'\u2561', file = sys.stderr) for frame in stack: print(u"\u2502 %*s \u2502" % (-width, frame), file = sys.stderr) print(u'\u2514' + u'\u2500' * (width + 2) + u'\u2518', file = sys.stderr) print(file=sys.stderr) cur = self.conn.cursor() try: if expr: cur.execute(expr, args) yield cur finally: cur.close()
