Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/jsonpickle/handlers.py: 51%
136 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:20 +0000
1"""
2Custom handlers may be created to handle other objects. Each custom handler
3must derive from :class:`jsonpickle.handlers.BaseHandler` and
4implement ``flatten`` and ``restore``.
6A handler can be bound to other types by calling
7:func:`jsonpickle.handlers.register`.
9"""
10from __future__ import absolute_import, division, unicode_literals
12import array
13import copy
14import datetime
15import io
16import re
17import sys
18import threading
19import uuid
21from . import compat, util
24class Registry(object):
25 def __init__(self):
26 self._handlers = {}
27 self._base_handlers = {}
29 def get(self, cls_or_name, default=None):
30 """
31 :param cls_or_name: the type or its fully qualified name
32 :param default: default value, if a matching handler is not found
34 Looks up a handler by type reference or its fully
35 qualified name. If a direct match
36 is not found, the search is performed over all
37 handlers registered with base=True.
38 """
39 handler = self._handlers.get(cls_or_name)
40 # attempt to find a base class
41 if handler is None and util.is_type(cls_or_name):
42 for cls, base_handler in self._base_handlers.items():
43 if issubclass(cls_or_name, cls):
44 return base_handler
45 return default if handler is None else handler
47 def register(self, cls, handler=None, base=False):
48 """Register the a custom handler for a class
50 :param cls: The custom object class to handle
51 :param handler: The custom handler class (if
52 None, a decorator wrapper is returned)
53 :param base: Indicates whether the handler should
54 be registered for all subclasses
56 This function can be also used as a decorator
57 by omitting the `handler` argument::
59 @jsonpickle.handlers.register(Foo, base=True)
60 class FooHandler(jsonpickle.handlers.BaseHandler):
61 pass
63 """
64 if handler is None:
66 def _register(handler_cls):
67 self.register(cls, handler=handler_cls, base=base)
68 return handler_cls
70 return _register
71 if not util.is_type(cls):
72 raise TypeError('{!r} is not a class/type'.format(cls))
73 # store both the name and the actual type for the ugly cases like
74 # _sre.SRE_Pattern that cannot be loaded back directly
75 self._handlers[util.importable_name(cls)] = self._handlers[cls] = handler
76 if base:
77 # only store the actual type for subclass checking
78 self._base_handlers[cls] = handler
80 def unregister(self, cls):
81 self._handlers.pop(cls, None)
82 self._handlers.pop(util.importable_name(cls), None)
83 self._base_handlers.pop(cls, None)
86registry = Registry()
87register = registry.register
88unregister = registry.unregister
89get = registry.get
92class BaseHandler(object):
93 def __init__(self, context):
94 """
95 Initialize a new handler to handle a registered type.
97 :Parameters:
98 - `context`: reference to pickler/unpickler
100 """
101 self.context = context
103 def flatten(self, obj, data):
104 """
105 Flatten `obj` into a json-friendly form and write result to `data`.
107 :param object obj: The object to be serialized.
108 :param dict data: A partially filled dictionary which will contain the
109 json-friendly representation of `obj` once this method has
110 finished.
111 """
112 raise NotImplementedError('You must implement flatten() in %s' % self.__class__)
114 def restore(self, obj):
115 """
116 Restore an object of the registered type from the json-friendly
117 representation `obj` and return it.
118 """
119 raise NotImplementedError('You must implement restore() in %s' % self.__class__)
121 @classmethod
122 def handles(self, cls):
123 """
124 Register this handler for the given class. Suitable as a decorator,
125 e.g.::
127 @MyCustomHandler.handles
128 class MyCustomClass:
129 def __reduce__(self):
130 ...
131 """
132 registry.register(cls, self)
133 return cls
135 def __call__(self, context):
136 """This permits registering either Handler instances or classes
138 :Parameters:
139 - `context`: reference to pickler/unpickler
140 """
141 self.context = context
142 return self
145class ArrayHandler(BaseHandler):
146 """Flatten and restore array.array objects"""
148 def flatten(self, obj, data):
149 data['typecode'] = obj.typecode
150 data['values'] = self.context.flatten(obj.tolist(), reset=False)
151 return data
153 def restore(self, data):
154 typecode = data['typecode']
155 values = self.context.restore(data['values'], reset=False)
156 if typecode == 'c':
157 values = [bytes(x) for x in values]
158 return array.array(typecode, values)
161ArrayHandler.handles(array.array)
164class DatetimeHandler(BaseHandler):
166 """Custom handler for datetime objects
168 Datetime objects use __reduce__, and they generate binary strings encoding
169 the payload. This handler encodes that payload to reconstruct the
170 object.
172 """
174 def flatten(self, obj, data):
175 pickler = self.context
176 if not pickler.unpicklable:
177 if hasattr(obj, 'isoformat'):
178 result = obj.isoformat()
179 else:
180 result = compat.ustr(obj)
181 return result
182 cls, args = obj.__reduce__()
183 flatten = pickler.flatten
184 payload = util.b64encode(args[0])
185 args = [payload] + [flatten(i, reset=False) for i in args[1:]]
186 data['__reduce__'] = (flatten(cls, reset=False), args)
187 return data
189 def restore(self, data):
190 cls, args = data['__reduce__']
191 unpickler = self.context
192 restore = unpickler.restore
193 cls = restore(cls, reset=False)
194 value = util.b64decode(args[0])
195 params = (value,) + tuple([restore(i, reset=False) for i in args[1:]])
196 return cls.__new__(cls, *params)
199DatetimeHandler.handles(datetime.datetime)
200DatetimeHandler.handles(datetime.date)
201DatetimeHandler.handles(datetime.time)
204class RegexHandler(BaseHandler):
205 """Flatten _sre.SRE_Pattern (compiled regex) objects"""
207 def flatten(self, obj, data):
208 data['pattern'] = obj.pattern
209 return data
211 def restore(self, data):
212 return re.compile(data['pattern'])
215RegexHandler.handles(type(re.compile('')))
218class QueueHandler(BaseHandler):
219 """Opaquely serializes Queue objects
221 Queues contains mutex and condition variables which cannot be serialized.
222 Construct a new Queue instance when restoring.
224 """
226 def flatten(self, obj, data):
227 return data
229 def restore(self, data):
230 return compat.queue.Queue()
233QueueHandler.handles(compat.queue.Queue)
236class CloneFactory(object):
237 """Serialization proxy for collections.defaultdict's default_factory"""
239 def __init__(self, exemplar):
240 self.exemplar = exemplar
242 def __call__(self, clone=copy.copy):
243 """Create new instances by making copies of the provided exemplar"""
244 return clone(self.exemplar)
246 def __repr__(self):
247 return '<CloneFactory object at 0x{:x} ({})>'.format(id(self), self.exemplar)
250class UUIDHandler(BaseHandler):
251 """Serialize uuid.UUID objects"""
253 def flatten(self, obj, data):
254 data['hex'] = obj.hex
255 return data
257 def restore(self, data):
258 return uuid.UUID(data['hex'])
261UUIDHandler.handles(uuid.UUID)
264class LockHandler(BaseHandler):
265 """Serialize threading.Lock objects"""
267 def flatten(self, obj, data):
268 data['locked'] = obj.locked()
269 return data
271 def restore(self, data):
272 lock = threading.Lock()
273 if data.get('locked', False):
274 lock.acquire()
275 return lock
278_lock = threading.Lock()
279LockHandler.handles(_lock.__class__)
282class TextIOHandler(BaseHandler):
283 """Serialize file descriptors as None because we cannot roundtrip"""
285 def flatten(self, obj, data):
286 return None
288 def restore(self, data):
289 """Restore should never get called because flatten() returns None"""
290 raise AssertionError('Restoring IO.TextIOHandler is not supported')
293if sys.version_info >= (3, 8):
294 TextIOHandler.handles(io.TextIOWrapper)