5648901703

pending completion

Build # 5648901703

Build Type

push

github-actions

Committed by

web-flow

Commit Message

refactor: support storages that encode as binary strings (#17)

Co-authored-by: jtbr <jtbr@users.noreply.github.com>

Run Details

6 of 6 new or added lines in 1 file covered. (100.0%)

84 of 88 relevant lines covered (95.45%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.0

/tinydb_serialization/__init__.py

from abc import ABC, abstractmethod
from copy import deepcopy

from tinydb import TinyDB
from tinydb.middlewares import Middleware

iteritems = getattr(dict, 'iteritems', dict.items)
itervalues = getattr(dict, 'itervalues', dict.values)


class Serializer(ABC):
    """
    The abstract base class for Serializers.
    Allows TinyDB to handle arbitrary objects by running them through a list
    of registerd serializers.
    Every serializer has to tell which class it can handle.
    """

    @property
    @abstractmethod
    def OBJ_CLASS(self):
        raise NotImplementedError('To be overriden!')

    @abstractmethod
    def encode(self, obj):
        """
        Encode an object.
        :param obj:
        :return:
        :rtype: str
        """
        raise NotImplementedError('To be overriden!')

    @abstractmethod
    def decode(self, s):
        """
        Decode an object.
        :param s:
        :type s: str
        :return:
        """
        raise NotImplementedError('To be overriden!')


def _enumerate_element(element):
    """
    Make an element enumerable.

    For dicts: return an iterator over the items (key, value).
    For lists/tuples: return an iterator over (index, item)
    """

    if isinstance(element, dict):
        return iteritems(element)
    else:
        return enumerate(element)


def _decode_deep(element, serializer, tag):
    """
    Recursively decode an element.

    Takes into account elements in nested dicts, lists and tuples
    """

    for key, value in _enumerate_element(element):
        try:
            typed_tag = tag
            if type(value) == bytes:
                typed_tag = tag.encode()
            if value.startswith(typed_tag):
                encoded = value[len(typed_tag):]
                element[key] = serializer.decode(encoded)

        except AttributeError:
            # Not a string
            if isinstance(value, (dict, list, tuple)):
                _decode_deep(value, serializer, tag)


def _encode_deep(element, serializer, tag, obj_class):
    """
    Recursively encode an element.

    Takes into account elements in nested dicts, lists and tuples
    """

    for key, value in _enumerate_element(element):
        if isinstance(value, obj_class):
            encoded = serializer.encode(value)
            element[key] = (tag if type(encoded) == str else tag.encode()) + encoded

        elif isinstance(value, (dict, list, tuple)):
            _encode_deep(value, serializer, tag, obj_class)


def has_encodable(element, obj_class):
    """
    Check whether the element in question has an encodable item.
    """

    found_encodable = False

    for key, value in _enumerate_element(element):
        if isinstance(value, (dict, list, tuple)):
            found_encodable |= has_encodable(value, obj_class)
        else:
            found_encodable |= isinstance(value, obj_class)

    return found_encodable


class SerializationMiddleware(Middleware):
    """
    Provide custom serialization for TinyDB.
    This middleware allows users of TinyDB to register custom serializations.
    The serialized data will be passed to the wrapped storage and data that
    is read from the storage will be deserialized.
    """

    def __init__(self, storage_cls=TinyDB.default_storage_class):
        super(SerializationMiddleware, self).__init__(storage_cls)

        self._serializers = {}

    def register_serializer(self, serializer, name):
        """
        Register a new Serializer.
        When reading from/writing to the underlying storage, TinyDB
        will run all objects through the list of registered serializers
        allowing each one to handle objects it recognizes.
        .. note:: The name has to be unique among this database instance.
                  Re-using the same name will overwrite the old serializer.
                  Also, registering a serializer will be reflected in all
                  tables when reading/writing them.
        :param serializer: an instance of the serializer
        :type serializer: tinydb.serialize.Serializer
        """
        self._serializers[name] = serializer

    def read(self):
        data = self.storage.read()

        if data is None:
            return None

        for serializer_name in self._serializers:
            serializer = self._serializers[serializer_name]
            tag = '{{{0}}}:'.format(serializer_name)  # E.g:'{TinyDate}:'

            for table_name in data:
                table = data[table_name]

                for eid in table:
                    _decode_deep(data[table_name][eid], serializer, tag)

        return data

    def write(self, data):
        # We only make a copy of the data if any serializer would overwrite
        # existing data.
        data_copied = False

        for serializer_name in self._serializers:
            # If no serializers are registered, this code will just look up
            # the serializer list and continue. But if there are serializers,
            # the inner loop will run very often.
            # For that reason, the lookup of the serialized class is pulled
            # out into the outer loop:

            serializer = self._serializers[serializer_name]
            obj_class = serializer.OBJ_CLASS
            tag = '{{{0}}}:'.format(serializer_name)

            for table_name in data:
                table = data[table_name]

                for eid in table:
                    # Before writing, copy data if we haven't already.
                    if not data_copied and has_encodable(data[table_name][eid],
                                                         obj_class):
                        data = deepcopy(data)
                        data_copied = True

                    item = data[table_name][eid]
                    _encode_deep(item, serializer, tag, obj_class)

        self.storage.write(data)

1	from abc import ABC, abstractmethod	1✔
2	from copy import deepcopy	1✔
3
4	from tinydb import TinyDB	1✔
5	from tinydb.middlewares import Middleware	1✔
6
7	iteritems = getattr(dict, 'iteritems', dict.items)	1✔
8	itervalues = getattr(dict, 'itervalues', dict.values)	1✔
9
10
11	class Serializer(ABC):	1✔
12	"""
13	The abstract base class for Serializers.
14	Allows TinyDB to handle arbitrary objects by running them through a list
15	of registerd serializers.
16	Every serializer has to tell which class it can handle.
17	"""
18
19	@property	1✔
20	@abstractmethod	1✔
21	def OBJ_CLASS(self):	1✔
22	raise NotImplementedError('To be overriden!')	×
23
24	@abstractmethod	1✔
25	def encode(self, obj):	1✔
26	"""
27	Encode an object.
28	:param obj:
29	:return:
30	:rtype: str
31	"""
32	raise NotImplementedError('To be overriden!')	×
33
34	@abstractmethod	1✔
35	def decode(self, s):	1✔
36	"""
37	Decode an object.
38	:param s:
39	:type s: str
40	:return:
41	"""
42	raise NotImplementedError('To be overriden!')	×
43
44
45	def _enumerate_element(element):	1✔
46	"""
47	Make an element enumerable.
48
49	For dicts: return an iterator over the items (key, value).
50	For lists/tuples: return an iterator over (index, item)
51	"""
52
53	if isinstance(element, dict):	1✔
54	return iteritems(element)	1✔
55	else:
56	return enumerate(element)	1✔
57
58
59	def _decode_deep(element, serializer, tag):	1✔
60	"""
61	Recursively decode an element.
62
63	Takes into account elements in nested dicts, lists and tuples
64	"""
65
66	for key, value in _enumerate_element(element):	1✔
67	try:	1✔
68	typed_tag = tag	1✔
69	if type(value) == bytes:	1✔
70	typed_tag = tag.encode()	×
71	if value.startswith(typed_tag):	1✔
72	encoded = value[len(typed_tag):]	1✔
73	element[key] = serializer.decode(encoded)	1✔
74
75	except AttributeError:	1✔
76	# Not a string
77	if isinstance(value, (dict, list, tuple)):	1✔
78	_decode_deep(value, serializer, tag)	1✔
79
80
81	def _encode_deep(element, serializer, tag, obj_class):	1✔
82	"""
83	Recursively encode an element.
84
85	Takes into account elements in nested dicts, lists and tuples
86	"""
87
88	for key, value in _enumerate_element(element):	1✔
89	if isinstance(value, obj_class):	1✔
90	encoded = serializer.encode(value)	1✔
91	element[key] = (tag if type(encoded) == str else tag.encode()) + encoded	1✔
92
93	elif isinstance(value, (dict, list, tuple)):	1✔
94	_encode_deep(value, serializer, tag, obj_class)	1✔
95
96
97	def has_encodable(element, obj_class):	1✔
98	"""
99	Check whether the element in question has an encodable item.
100	"""
101
102	found_encodable = False	1✔
103
104	for key, value in _enumerate_element(element):	1✔
105	if isinstance(value, (dict, list, tuple)):	1✔
106	found_encodable \|= has_encodable(value, obj_class)	1✔
107	else:
108	found_encodable \|= isinstance(value, obj_class)	1✔
109
110	return found_encodable	1✔
111
112
113	class SerializationMiddleware(Middleware):	1✔
114	"""
115	Provide custom serialization for TinyDB.
116	This middleware allows users of TinyDB to register custom serializations.
117	The serialized data will be passed to the wrapped storage and data that
118	is read from the storage will be deserialized.
119	"""
120
121	def __init__(self, storage_cls=TinyDB.default_storage_class):	1✔
122	super(SerializationMiddleware, self).__init__(storage_cls)	1✔
123
124	self._serializers = {}	1✔
125
126	def register_serializer(self, serializer, name):	1✔
127	"""
128	Register a new Serializer.
129	When reading from/writing to the underlying storage, TinyDB
130	will run all objects through the list of registered serializers
131	allowing each one to handle objects it recognizes.
132	.. note:: The name has to be unique among this database instance.
133	Re-using the same name will overwrite the old serializer.
134	Also, registering a serializer will be reflected in all
135	tables when reading/writing them.
136	:param serializer: an instance of the serializer
137	:type serializer: tinydb.serialize.Serializer
138	"""
139	self._serializers[name] = serializer	1✔
140
141	def read(self):	1✔
142	data = self.storage.read()	1✔
143
144	if data is None:	1✔
145	return None	1✔
146
147	for serializer_name in self._serializers:	1✔
148	serializer = self._serializers[serializer_name]	1✔
149	tag = '{{{0}}}:'.format(serializer_name) # E.g:'{TinyDate}:'	1✔
150
151	for table_name in data:	1✔
152	table = data[table_name]	1✔
153
154	for eid in table:	1✔
155	_decode_deep(data[table_name][eid], serializer, tag)	1✔
156
157	return data	1✔
158
159	def write(self, data):	1✔
160	# We only make a copy of the data if any serializer would overwrite
161	# existing data.
162	data_copied = False	1✔
163
164	for serializer_name in self._serializers:	1✔
165	# If no serializers are registered, this code will just look up
166	# the serializer list and continue. But if there are serializers,
167	# the inner loop will run very often.
168	# For that reason, the lookup of the serialized class is pulled
169	# out into the outer loop:
170
171	serializer = self._serializers[serializer_name]	1✔
172	obj_class = serializer.OBJ_CLASS	1✔
173	tag = '{{{0}}}:'.format(serializer_name)	1✔
174
175	for table_name in data:	1✔
176	table = data[table_name]	1✔
177
178	for eid in table:	1✔
179	# Before writing, copy data if we haven't already.
180	if not data_copied and has_encodable(data[table_name][eid],	1✔
181	obj_class):
182	data = deepcopy(data)	1✔
183	data_copied = True	1✔
184
185	item = data[table_name][eid]	1✔
186	_encode_deep(item, serializer, tag, obj_class)	1✔
187
188	self.storage.write(data)	1✔

msiemens / tinydb-serialization / 5648901703

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous