我们从Python开源项目中，提取了以下20个代码示例，用于说明如何使用sqlite3.register_converter()。
def _updateProgramList(self, programList, channel): # todo workaround service.py 'forgets' the adapter and convert set in _initialize.. wtf?! sqlite3.register_adapter(datetime.datetime, self.adapt_datetime) sqlite3.register_converter('timestamp', self.convert_datetime) c = self.conn.cursor() c.execute('DELETE FROM programs WHERE source=? AND channel=? ', [self.source.KEY, channel.id]) updatesId = 1 #TODO why? for program in programList: c.execute( 'INSERT OR REPLACE INTO programs(channel, title, sub_title, start_date, end_date, description, categories, image_large, image_small, season, episode, is_movie, language, source, updates_id) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)', [channel.id, program.title, program.sub_title, program.startDate, program.endDate, program.description, program.categories, program.imageLarge, program.imageSmall, program.season, program.episode, program.is_movie, program.language, self.source.KEY, updatesId]) self.conn.commit()
def __init__(self, path: str) -> None: self.path = path if os.path.isfile(self.path): self._conn = sqlite3.connect( self.path, detect_types=sqlite3.PARSE_DECLTYPES, isolation_level="DEFERRED") self._conn.create_function("gen_salt", 0, lambda: secure_string(8)) self._cur = self._conn.cursor() self._cur.arraysize = 20 self._cur.executescript("""\ PRAGMA foreign_keys = ON; """) else: self._conn = None self._cur = None # Create adapter from python boolean to sqlite integer. sqlite3.register_adapter(bool, int) sqlite3.register_converter("BOOL", lambda x: bool(int(x)))
def setUp(self): self.con = sqlite.connect(":memory:", detect_types=sqlite.PARSE_COLNAMES) sqlite.register_converter("bin", BinaryConverterTests.convert)
def __init__(self, name='history'): ''' Constructor ''' # counts the number of turns in a game self.counter = 1 # refer to the next turn self.c_turn = 2 # names for saving etc. self.name = name self.game = 'default_setting' #create database self.connection = sqlite3.connect(self.name+'.db', detect_types=sqlite3.PARSE_DECLTYPES) self.cursor = self.connection.cursor() # define array data type sqlite3.register_adapter(np.ndarray, self.__adapt_array) sqlite3.register_converter('ARRAY', self.__convert_array) # create table self.cursor.execute('''CREATE TABLE IF NOT EXISTS games ( game TEXT UNIQUE, turns INTEGER, error ARRAY)''') self.cursor.execute('''CREATE TABLE IF NOT EXISTS default_setting ( number INTEGER UNIQUE, player TEXT, state ARRAY, action ARRAY, reward INTEGER)''') # makes sure to save atexit.register(self.__del__)
def registerType(cls: type, pyToDb: Callable[[ActiveRecord], T], dbToPy: Callable[[T], ActiveRecord]): ActiveRecord.registeredTypes.add(cls.__name__) sqlite3.register_adapter(cls, pyToDb) sqlite3.register_converter(cls.__name__, dbToPy) # Got types are registered in their class files. Built-in types are registered here
def _invokeAndBlockForResult(self, method, *args): sqlite3.register_adapter(datetime.datetime, self.adapt_datetime) sqlite3.register_converter('timestamp', self.convert_datetime) event = [method, None] event.extend(args) self.eventQueue.append(event) self.event.set() while not method.__name__ in self.eventResults: time.sleep(0.1) result = self.eventResults.get(method.__name__) del self.eventResults[method.__name__] return result
def _sqlite_adapters(): def adapt_ieee(eui64): return repr(eui64) sqlite3.register_adapter(t.EmberEUI64, adapt_ieee) def convert_ieee(s): l = [t.uint8_t(p, base=16) for p in s.split(b':')] return t.EmberEUI64(l) sqlite3.register_converter("ieee", convert_ieee)
def connect(): sqlite3.register_adapter(uuid.UUID, adapt_uuid) sqlite3.register_converter('uuid', convert_uuid) sqlite3.register_converter('timeuuid', convert_uuid) sqlite3.register_adapter(list, adapt_strset) sqlite3.register_converter('strset', convert_strset) c.sqlite3 = sqlite3.connect(mqeconfig.SQLITE3_DATABASE, timeout=TIMEOUT, detect_types=sqlite3.PARSE_DECLTYPES) c.sqlite3.row_factory = dict_factory with closing_cursor() as cur: cur.execute("""PRAGMA case_sensitive_like = true;""")
def _initialize(self, cancel_requested_callback): sqlite3.register_adapter(datetime.datetime, self.adapt_datetime) sqlite3.register_converter('timestamp', self.convert_datetime) self.alreadyTriedUnlinking = False while True: if cancel_requested_callback is not None and cancel_requested_callback(): break try: self.conn = sqlite3.connect(self.databasePath, detect_types=sqlite3.PARSE_DECLTYPES) self.conn.execute('PRAGMA foreign_keys = ON') self.conn.row_factory = sqlite3.Row # create and drop dummy table to check if database is locked c = self.conn.cursor() c.execute('CREATE TABLE IF NOT EXISTS database_lock_check(id TEXT PRIMARY KEY)') c.execute('DROP TABLE database_lock_check') c.close() self._createTables() self.settingsChanged = self._wasSettingsChanged(ADDON) break except sqlite3.OperationalError: if cancel_requested_callback is None: xbmc.log('[script.tvguide.fullscreen] Database is locked, bailing out...', xbmc.LOGDEBUG) break else: # ignore 'database is locked' xbmc.log('[script.tvguide.fullscreen] Database is locked, retrying...', xbmc.LOGDEBUG) except sqlite3.DatabaseError: self.conn = None if self.alreadyTriedUnlinking: xbmc.log('[script.tvguide.fullscreen] Database is broken and unlink() failed', xbmc.LOGDEBUG) break else: try: os.unlink(self.databasePath) except OSError: pass self.alreadyTriedUnlinking = True xbmcgui.Dialog().ok(ADDON.getAddonInfo('name'), strings(DATABASE_SCHEMA_ERROR_1), strings(DATABASE_SCHEMA_ERROR_2), strings(DATABASE_SCHEMA_ERROR_3)) return self.conn is not None
def __init__(self, filename=None, clean_n_entries=30): """ :param filename: path to database file :param clean_n_entries: during cleaning delete n % entries. """ self.clean_n_entries = clean_n_entries import sqlite3 # register numpy array conversion functions # uses "np_array" type in sql tables def adapt_array(arr): out = BytesIO() np.savez_compressed(out, offsets=arr) out.seek(0) return out.read() def convert_array(text): out = BytesIO(text) out.seek(0) npz = np.load(out) arr = npz['offsets'] npz.close() return arr # Converts np.array to TEXT when inserting sqlite3.register_adapter(np.ndarray, adapt_array) # Converts TEXT to np.array when selecting sqlite3.register_converter("NPARRAY", convert_array) self._database = sqlite3.connect(filename if filename is not None else ":memory:", detect_types=sqlite3.PARSE_DECLTYPES, timeout=1000*1000, isolation_level=None) self.filename = filename try: cursor = self._database.execute("select num from version") row = cursor.fetchone() if not row: self.db_version = TrajectoryInfoCache.DB_VERSION version = self.db_version else: version = row[0] if version != TrajectoryInfoCache.DB_VERSION: # drop old db? or try to convert? self._create_new_db() except sqlite3.OperationalError as e: if "no such table" in str(e): self._create_new_db() self.db_version = TrajectoryInfoCache.DB_VERSION except sqlite3.DatabaseError: bak = filename + ".bak" warnings.warn("TrajInfo database corrupted. Backing up file to %s and start with new one." % bak) self._database.close() import shutil shutil.move(filename, bak) SqliteDB.__init__(self, filename)
