How to use the tinydb.storages.Storage function in tinydb

To help you get started, we’ve selected a few tinydb examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github msiemens / tinydb / tests / test_storages.py View on Github external
:type tmpdir: py._path.local.LocalPath
    """

    try:
        import yaml
    except ImportError:
        return pytest.skip('PyYAML not installed')

    def represent_doc(dumper, data):
        # Represent `Document` objects as their dict's string representation
        # which PyYAML understands
        return dumper.represent_data(dict(data))

    yaml.add_representer(Document, represent_doc)

    class YAMLStorage(Storage):
        def __init__(self, filename):
            self.filename = filename
            touch(filename, False)

        def read(self):
            with open(self.filename) as handle:
                data = yaml.safe_load(handle.read())
                return data

        def write(self, data):
            with open(self.filename, 'w') as handle:
                yaml.dump(data, handle)

        def close(self):
            pass
github msiemens / tinydb / tests / test_storages.py View on Github external
def test_read_once():
    count = 0

    # noinspection PyAbstractClass
    class MyStorage(Storage):
        def __init__(self):
            self.memory = None

        def read(self):
            nonlocal count
            count += 1

            return self.memory

        def write(self, data):
            self.memory = data

    with TinyDB(storage=MyStorage) as db:
        assert count == 0

        db.table()
github msiemens / tinydb / tinydb / storages.py View on Github external
# File is empty
            return None
        else:
            self._handle.seek(0)
            return json.load(self._handle)

    def write(self, data: Dict[str, Dict[str, Any]]):
        self._handle.seek(0)
        serialized = json.dumps(data, **self.kwargs)
        self._handle.write(serialized)
        self._handle.flush()
        os.fsync(self._handle.fileno())
        self._handle.truncate()


class MemoryStorage(Storage):
    """
    Store the data as JSON in memory.
    """

    def __init__(self):
        """
        Create a new instance.
        """

        super().__init__()
        self.memory = None

    def read(self) -> Optional[Dict[str, Dict[str, Any]]]:
        return self.memory

    def write(self, data: Dict[str, Dict[str, Any]]):
github msiemens / tinydb / tinydb / storages.py View on Github external
Any kind of serialization should go here.

        :param data: The current state of the database.
        """

        raise NotImplementedError('To be overridden!')

    def close(self) -> None:
        """
        Optional: Close open file handles, etc.
        """

        pass


class JSONStorage(Storage):
    """
    Store the data in a JSON file.
    """

    def __init__(self, path: str, create_dirs=False, encoding=None, **kwargs):
        """
        Create a new instance.

        Also creates the storage file, if it doesn't exist.

        :param path: Where to store the JSON data.
        """

        super().__init__()
        touch(path, create_dirs=create_dirs)  # Create file if not exists
        self.kwargs = kwargs
github mbr / tinydb-git / tinydb_git / __init__.py View on Github external
import time

from dulwich.repo import Repo
from dulwich.objects import Commit, Tree, Blob
from tinydb.storages import Storage

__version__ = '0.2.dev1'


class GitStorage(Storage):
    def __init__(self, repo_path, branch, filename):
        self.branch = branch
        self.filename = filename

        if b'/' in filename:
            raise NotImplementedError(
                'Currently, subdir support is not implemented. Annoy the '
                'author on github to get it done.')

        self.repo = Repo(repo_path)

    @property
    def _refname(self):
        return b'refs/heads/' + self.branch

    def read(self):