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_access_storage():
assert isinstance(TinyDB(storage=MemoryStorage).storage,
MemoryStorage)
assert isinstance(TinyDB(storage=CachingMiddleware(MemoryStorage)).storage,
CachingMiddleware)
def test_nested():
storage = CachingMiddleware(MemoryStorage)
storage() # Initialization
# Write contents
storage.write(doc)
# Verify contents
assert doc == storage.read()
def db():
return TinyDB(storage=MemoryStorage).table('table')
def test_in_memory():
# Write contents
storage = MemoryStorage()
storage.write(doc)
# Verify contents
assert doc == storage.read()
# Test case for #21
other = MemoryStorage()
other.write({})
assert other.read() != storage.read()
def __init__(self, url):
"""
Construct method
"""
self._db = TinyDB(storage=MemoryStorage)
if url and re.match(r"(http://)?(\d+\.\d+\.\d+\.\d+:\d+)", url):
if '://' not in url:
url = 'http://' + url
else:
url = url
self._url = url
self.load()
else:
logger.error('Atx server addr error')
self.load()
def __init__(self):
self._db = TinyDB(storage=MemoryStorage)
self._new_channel = asyncio.Condition()
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from tinydb import TinyDB
from tinydb.storages import MemoryStorage, JSONStorage
class DB(TinyDB):
# To modify the default storage for all TinyDB instances
TinyDB.DEFAULT_STORAGE = MemoryStorage
pass
class MemStroage(MemoryStorage):
pass
class JsonStorage(JSONStorage):
pass
def __init__(self, url=None, token=None):
"""
Construct method
"""
self._db = TinyDB(storage=MemoryStorage)
self._url = url
self._token = token
def load_datasets(dataset_filenames):
"""
Create a PickelableTinyDB with the data from a list of filenames.
Parameters
----------
dataset_filenames : [str]
List of filenames to load as datasets
Returns
-------
PickleableTinyDB
"""
ds_database = PickleableTinyDB(storage=MemoryStorage)
for fname in dataset_filenames:
with open(fname) as file_:
try:
d = json.load(file_)
check_dataset(d)
ds_database.insert(clean_dataset(d))
except ValueError as e:
raise ValueError('JSON Error in {}: {}'.format(fname, e))
except DatasetError as e:
raise DatasetError('Dataset Error in {}: {}'.format(fname, e))
return ds_database