5648906116

pending completion

Build # 5648906116

Build Type

push

github-actions

Committed by

web-flow

Commit Message

fix: "Type[type]" is incompatible with "Type[property]" (#14)

* update .gitignore

* fix misspelled  words

* fix: "Type[type]" is incompatible with "Type[property]"

Run Details

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

82 of 85 relevant lines covered (96.47%)

1.93 hits per line

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

96.1

/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 registered serializers.
    Every serializer has to tell which class it can handle.
    """

    OBJ_CLASS: object

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

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


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_encodeable(element, obj_class):
    """
    Check whether the element in question has an encodeable item.
    """

    found_encodeable = False

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

    return found_encodeable


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

msiemens / tinydb-serialization / 5648906116

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