Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/jsonpickle/handlers.py: 56%

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``. 

5 

6A handler can be bound to other types by calling 

7:func:`jsonpickle.handlers.register`. 

8 

9""" 

10 

11import array 

12import copy 

13import datetime 

14import io 

15import queue 

16import re 

17import threading 

18import uuid 

19from typing import Any, Callable, Dict, NoReturn, Optional, Type, TypeVar, Union 

20 

21from . import util 

22 

23T = TypeVar("T") 

24 

25# Nb. we can't import the below types directly from pickler/unpickler 

26# without introducing a circular import dependency. 

27ContextType = Union[ # type: ignore[valid-type] 

28 TypeVar("Pickler", bound="Pickler"), # noqa: F821 

29 TypeVar("Unpickler", bound="Unpickler"), # noqa: F821 

30] 

31HandlerType = Type[Any] 

32KeyType = Union[Type[Any], str] 

33HandlerReturn = Optional[Union[dict[str, Any], str]] 

34DateTime = Union[datetime.datetime, datetime.date, datetime.time] 

35 

36 

37class Registry: 

38 def __init__(self) -> None: 

39 self._handlers = {} 

40 self._base_handlers = {} 

41 

42 def get(self, cls_or_name: type, default: Optional[Any] = None) -> Any: 

43 """ 

44 :param cls_or_name: the type or its fully qualified name 

45 :param default: default value, if a matching handler is not found 

46 

47 Looks up a handler by type reference or its fully 

48 qualified name. If a direct match 

49 is not found, the search is performed over all 

50 handlers registered with base=True. 

51 """ 

52 handler = self._handlers.get(cls_or_name) 

53 # attempt to find a base class 

54 if handler is None and util._is_type(cls_or_name): 

55 for cls, base_handler in self._base_handlers.items(): 

56 if issubclass(cls_or_name, cls): 

57 return base_handler 

58 return default if handler is None else handler 

59 

60 def register( 

61 self, cls: Type[Any], handler: Optional[KeyType] = None, base: bool = False 

62 ) -> Optional[Callable[[HandlerType], HandlerType]]: 

63 """Register the a custom handler for a class 

64 

65 :param cls: The custom object class to handle 

66 :param handler: The custom handler class (if 

67 None, a decorator wrapper is returned) 

68 :param base: Indicates whether the handler should 

69 be registered for all subclasses 

70 

71 This function can be also used as a decorator 

72 by omitting the `handler` argument:: 

73 

74 @jsonpickle.handlers.register(Foo, base=True) 

75 class FooHandler(jsonpickle.handlers.BaseHandler): 

76 pass 

77 

78 """ 

79 if handler is None: 

80 

81 def _register(handler_cls: Type[Any]) -> Type[Any]: 

82 self.register(cls, handler=handler_cls, base=base) 

83 return handler_cls 

84 

85 return _register 

86 if not util._is_type(cls): 

87 raise TypeError(f"{cls!r} is not a class/type") 

88 # store both the name and the actual type for the ugly cases like 

89 # _sre.SRE_Pattern that cannot be loaded back directly 

90 self._handlers[util.importable_name(cls)] = self._handlers[cls] = handler 

91 if base: 

92 # only store the actual type for subclass checking 

93 self._base_handlers[cls] = handler 

94 

95 def unregister(self, cls: Type[Any]) -> None: 

96 self._handlers.pop(cls, None) 

97 self._handlers.pop(util.importable_name(cls), None) 

98 self._base_handlers.pop(cls, None) 

99 

100 

101registry = Registry() 

102register = registry.register 

103unregister = registry.unregister 

104get = registry.get 

105 

106 

107class BaseHandler: 

108 def __init__(self, context: Any): 

109 """ 

110 Initialize a new handler to handle a registered type. 

111 

112 :Parameters: 

113 - `context`: reference to pickler/unpickler 

114 

115 """ 

116 self.context = context 

117 

118 def flatten(self, obj: Any, data: Dict[str, Any]) -> HandlerReturn: 

119 """ 

120 Flatten `obj` into a json-friendly form and write result to `data`. 

121 

122 :param object obj: The object to be serialized. 

123 :param dict data: A partially filled dictionary which will contain the 

124 json-friendly representation of `obj` once this method has 

125 finished. 

126 """ 

127 raise NotImplementedError("You must implement flatten() in %s" % self.__class__) 

128 

129 def restore(self, data: Dict[str, Any]) -> Any: 

130 """ 

131 Restore an object of the registered type from the json-friendly 

132 representation `obj` and return it. 

133 """ 

134 raise NotImplementedError("You must implement restore() in %s" % self.__class__) 

135 

136 @classmethod 

137 def handles(self, cls: Type[Any]) -> Type[Any]: 

138 """ 

139 Register this handler for the given class. Suitable as a decorator, 

140 e.g.:: 

141 

142 @MyCustomHandler.handles 

143 class MyCustomClass: 

144 def __reduce__(self): 

145 ... 

146 """ 

147 registry.register(cls, self) 

148 return cls 

149 

150 def __call__( 

151 self, context: ContextType # type: ignore[valid-type] 

152 ) -> "BaseHandler": 

153 """This permits registering either Handler instances or classes 

154 

155 :Parameters: 

156 - `context`: reference to pickler/unpickler 

157 """ 

158 self.context = context 

159 return self 

160 

161 

162class ArrayHandler(BaseHandler): 

163 """Flatten and restore array.array objects""" 

164 

165 def flatten(self, obj: array.array, data: Dict[str, Any]) -> HandlerReturn: # type: ignore[type-arg] 

166 data["typecode"] = obj.typecode 

167 data["values"] = self.context.flatten(obj.tolist(), reset=False) 

168 return data 

169 

170 def restore(self, data: Dict[str, Any]) -> array.array: # type: ignore[type-arg] 

171 typecode = data["typecode"] 

172 values = self.context.restore(data["values"], reset=False) 

173 if typecode == "c": 

174 values = [bytes(x) for x in values] 

175 return array.array(typecode, values) 

176 

177 

178ArrayHandler.handles(array.array) 

179 

180 

181class DatetimeHandler(BaseHandler): 

182 """Custom handler for datetime objects 

183 

184 Datetime objects use __reduce__, and they generate binary strings encoding 

185 the payload. This handler encodes that payload to reconstruct the 

186 object. 

187 

188 """ 

189 

190 def flatten(self, obj: DateTime, data: Dict[str, Any]) -> HandlerReturn: 

191 pickler = self.context 

192 if not pickler.unpicklable: 

193 if hasattr(obj, "isoformat"): 

194 result = obj.isoformat() 

195 else: 

196 result = str(obj) 

197 return result 

198 cls, args = obj.__reduce__() # type: ignore[misc] 

199 flatten = pickler.flatten 

200 payload = util.b64encode(args[0]) 

201 args = [payload] + [flatten(i, reset=False) for i in args[1:]] 

202 data["__reduce__"] = (flatten(cls, reset=False), args) 

203 return data 

204 

205 def restore(self, data: Dict[str, Any]) -> Any: 

206 cls, args = data["__reduce__"] 

207 unpickler = self.context 

208 restore = unpickler.restore 

209 cls = restore(cls, reset=False) 

210 value = util.b64decode(args[0]) 

211 params = (value,) + tuple([restore(i, reset=False) for i in args[1:]]) 

212 return cls.__new__(cls, *params) 

213 

214 

215DatetimeHandler.handles(datetime.datetime) 

216DatetimeHandler.handles(datetime.date) 

217DatetimeHandler.handles(datetime.time) 

218 

219 

220class RegexHandler(BaseHandler): 

221 """Flatten _sre.SRE_Pattern (compiled regex) objects""" 

222 

223 def flatten(self, obj: re.Pattern[str], data: Dict[str, Any]) -> HandlerReturn: 

224 data["pattern"] = obj.pattern 

225 return data 

226 

227 def restore(self, data: Dict[str, Any]) -> re.Pattern[str]: 

228 return re.compile(data["pattern"]) 

229 

230 

231RegexHandler.handles(type(re.compile(""))) 

232 

233 

234class QueueHandler(BaseHandler): 

235 """Opaquely serializes Queue objects 

236 

237 Queues contains mutex and condition variables which cannot be serialized. 

238 Construct a new Queue instance when restoring. 

239 

240 """ 

241 

242 def flatten(self, obj: queue.Queue[Any], data: Dict[str, Any]) -> HandlerReturn: 

243 return data 

244 

245 def restore(self, data: Dict[str, Any]) -> queue.Queue[Any]: 

246 return queue.Queue() 

247 

248 

249QueueHandler.handles(queue.Queue) 

250 

251 

252class CloneFactory: 

253 """Serialization proxy for collections.defaultdict's default_factory""" 

254 

255 def __init__(self, exemplar: Any) -> None: 

256 self.exemplar = exemplar 

257 

258 def __call__(self, clone: Callable[[Any], Any] = copy.copy) -> Any: 

259 """Create new instances by making copies of the provided exemplar""" 

260 return clone(self.exemplar) 

261 

262 def __repr__(self) -> str: 

263 return f"<CloneFactory object at 0x{id(self):x} ({self.exemplar})>" 

264 

265 

266class UUIDHandler(BaseHandler): 

267 """Serialize uuid.UUID objects""" 

268 

269 def flatten(self, obj: uuid.UUID, data: Dict[str, Any]) -> HandlerReturn: 

270 data["hex"] = obj.hex 

271 return data 

272 

273 def restore(self, data: Dict[str, Any]) -> uuid.UUID: 

274 return uuid.UUID(data["hex"]) 

275 

276 

277UUIDHandler.handles(uuid.UUID) 

278 

279 

280class LockHandler(BaseHandler): 

281 """Serialize threading.Lock objects""" 

282 

283 def flatten(self, obj: Any, data: dict[str, Any]) -> HandlerReturn: 

284 data["locked"] = obj.locked() 

285 return data 

286 

287 def restore(self, data: Dict[str, Any]) -> Any: 

288 lock = threading.Lock() 

289 if data.get("locked", False): 

290 lock.acquire() 

291 return lock 

292 

293 

294_lock = threading.Lock() 

295LockHandler.handles(_lock.__class__) 

296 

297 

298class TextIOHandler(BaseHandler): 

299 """Serialize file descriptors as None because we cannot roundtrip""" 

300 

301 def flatten(self, obj: io.TextIOBase, data: Dict[str, Any]) -> None: 

302 return None 

303 

304 def restore(self, data: Dict[str, Any]) -> NoReturn: 

305 """Restore should never get called because flatten() returns None""" 

306 raise AssertionError("Restoring IO.TextIOHandler is not supported") 

307 

308 

309TextIOHandler.handles(io.TextIOWrapper)