Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_migrate(self):
# adding manually items with plain text keys
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (b"abcd", sqlitedict.encode(u"abcd")))
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (b"xyz", sqlitedict.encode(24)))
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (u"+ěščřžýáíé%123456789úů".encode('utf8'), sqlitedict.encode(u"special")))
# migrating the DB
sqlite_migration.migrate(self.table_name, self.conn_old, self.conn_new)
# checking migrated DB via sqlitedict
d = sqlitedict.SqliteDict(filename=self.new_file, tablename=self.table_name, autocommit=True)
self.assertEqual(d[b"abcd"], u"abcd")
self.assertEqual(d[b"xyz"], 24)
# jquast: please note, previously when db[u"unicode"] was stored, it
# was returned as utf-8 encoded bytes. Going forward, it will store
# as u"unicode", so any dependent code must make its own further
# migration to re-encode all key-as-bytes as key-as-unicode.
self.assertEqual(d[u"+ěščřžýáíé%123456789úů".encode('utf8')], u"special")
d.terminate()
def test_migrate(self):
# adding manually items with plain text keys
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (b"abcd", sqlitedict.encode(u"abcd")))
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (b"xyz", sqlitedict.encode(24)))
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (u"+ěščřžýáíé%123456789úů".encode('utf8'), sqlitedict.encode(u"special")))
# migrating the DB
sqlite_migration.migrate(self.table_name, self.conn_old, self.conn_new)
# checking migrated DB via sqlitedict
d = sqlitedict.SqliteDict(filename=self.new_file, tablename=self.table_name, autocommit=True)
self.assertEqual(d[b"abcd"], u"abcd")
self.assertEqual(d[b"xyz"], 24)
# jquast: please note, previously when db[u"unicode"] was stored, it
# was returned as utf-8 encoded bytes. Going forward, it will store
# as u"unicode", so any dependent code must make its own further
# migration to re-encode all key-as-bytes as key-as-unicode.
self.assertEqual(d[u"+ěščřžýáíé%123456789úů".encode('utf8')], u"special")
d.terminate()
def setUp(self):
# create database of v1.2 values
self.filename = norm_file(
os.path.join(os.path.dirname(__file__),
'db', 'unmigrated.sqlite'))
self.db = sqlitedict.SqliteDict(filename=self.filename, flag='n')
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.db.tablename
keyval = (b"bytes", sqlitedict.encode(('some', 'value')))
self.db.conn.execute(ADD_ITEM, keyval)
def test_migrate(self):
# adding manually items with plain text keys
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (b"abcd", sqlitedict.encode(u"abcd")))
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (b"xyz", sqlitedict.encode(24)))
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % self.table_name
self.conn_old.execute(ADD_ITEM, (u"+ěščřžýáíé%123456789úů".encode('utf8'), sqlitedict.encode(u"special")))
# migrating the DB
sqlite_migration.migrate(self.table_name, self.conn_old, self.conn_new)
# checking migrated DB via sqlitedict
d = sqlitedict.SqliteDict(filename=self.new_file, tablename=self.table_name, autocommit=True)
self.assertEqual(d[b"abcd"], u"abcd")
self.assertEqual(d[b"xyz"], 24)
# jquast: please note, previously when db[u"unicode"] was stored, it
# was returned as utf-8 encoded bytes. Going forward, it will store
# as u"unicode", so any dependent code must make its own further
# migration to re-encode all key-as-bytes as key-as-unicode.
self.assertEqual(d[u"+ěščřžýáíé%123456789úů".encode('utf8')], u"special")
d.terminate()
def migrate(table_name, conn_old, conn_new):
# get all old keys
GET_KEYS = 'SELECT key FROM %s ORDER BY rowid' % table_name
old_keys = [key[0] for key in conn_old.select(GET_KEYS)]
for pos, key in enumerate(old_keys):
if pos % 100 == 0:
logger.info("converted %ith element of the DB", pos)
# get value for the given key
GET_ITEM = 'SELECT value FROM %s WHERE key = ?' % table_name
item = conn_old.select_one(GET_ITEM, (key,))
# save record with new key to new sqlite DB
ADD_ITEM = 'REPLACE INTO %s (key, value) VALUES (?,?)' % table_name
conn_new.execute(ADD_ITEM, (encode(key), item[0]))