Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _from_existing_db(db, complevel=6):
readonly = db.flags().get('readonly')
try:
db_info = db.open_db(b'info', create=False)
except lmdb.NotFoundError:
raise ValueError('Existing database is not a ds cache')
with db.begin(db_info, write=False) as tr:
version = tr.get(b'version', None)
if version is None:
raise ValueError('Missing format version field')
if version != FORMAT_VERSION:
raise ValueError("Unsupported on disk version: " + version.decode('utf8'))
zdict = tr.get(b'zdict', None)
dbs = SimpleNamespace(main=db,
info=db_info,
groups=db.open_db(b'groups', create=False),
ds=db.open_db(b'ds', create=False),
udata=db.open_db(b'udata', create=False))
try:
self.set_binlog(enable=False)
if binlog:
self.set_binlog(enable=True)
# try:
# self._semaphore = Semaphore(semaphore_path(name))
# except ExistentialError:
# pass
self._binlog = self._env.open_db(b'__binlog__', create=binlog)
self._binidx = self._env.open_db(b'__binidx__', create=binlog)
with self._env.begin(write=True) as txn:
if not txn.stat(db=self._binlog)['entries']:
dat = dumps({'txn': []}).encode()
txn.put(pack('>Q', 1), dat, db=self._binlog, append=False)
except lmdb.NotFoundError:
pass
except:
raise