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

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 inspect 

15import io 

16import queue 

17import re 

18import threading 

19import uuid 

20from collections.abc import Callable 

21from typing import TYPE_CHECKING, Any, NoReturn, TypeAlias, TypeVar 

22 

23from . import tags, util 

24 

25T = TypeVar("T") 

26 

27if TYPE_CHECKING: 

28 from .pickler import Pickler 

29 from .unpickler import Unpickler 

30 

31ContextType: TypeAlias = "Pickler | Unpickler" 

32RestoreType: TypeAlias = "Unpickler" 

33HandlerType: TypeAlias = type 

34KeyType: TypeAlias = type | str 

35HandlerReturn: TypeAlias = dict[str, Any] | str | None 

36DateTime: TypeAlias = datetime.date | datetime.time 

37 

38 

39class Registry: 

40 def __init__(self) -> None: 

41 self._handlers = {} 

42 self._base_handlers = {} 

43 

44 def get(self, cls_or_name: type, default: Any | None = None) -> Any: 

45 """ 

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

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

48 

49 Looks up a handler by type reference or its fully 

50 qualified name. If a direct match 

51 is not found, the search is performed over all 

52 handlers registered with base=True. 

53 """ 

54 handler = self._handlers.get(cls_or_name) 

55 # attempt to find a base class 

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

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

58 if issubclass(cls_or_name, cls): 

59 return base_handler 

60 return default if handler is None else handler 

61 

62 def register( 

63 self, cls: type, handler: KeyType | None = None, base: bool = False 

64 ) -> Callable[[HandlerType], HandlerType] | None: 

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

66 

67 :param cls: The custom object class to handle 

68 :param handler: The custom handler class (if 

69 None, a decorator wrapper is returned) 

70 :param base: Indicates whether the handler should 

71 be registered for all subclasses 

72 

73 This function can be also used as a decorator 

74 by omitting the `handler` argument:: 

75 

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

77 class FooHandler(jsonpickle.handlers.BaseHandler): 

78 pass 

79 

80 """ 

81 if handler is None: 

82 

83 def _register(handler_cls: HandlerType) -> HandlerType: 

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

85 return handler_cls 

86 

87 return _register 

88 if not util._is_type(cls): 

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

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

91 # _sre.SRE_Pattern that cannot be loaded back directly 

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

93 if base: 

94 # only store the actual type for subclass checking 

95 self._base_handlers[cls] = handler 

96 

97 def unregister(self, cls: type) -> None: 

98 self._handlers.pop(cls, None) 

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

100 self._base_handlers.pop(cls, None) 

101 

102 

103registry = Registry() 

104register = registry.register 

105unregister = registry.unregister 

106get = registry.get 

107 

108 

109def handler_accepts_handler_context(fn: Callable[..., Any]) -> bool: 

110 """ 

111 Check if the handler function has a handler_context parameter. 

112 """ 

113 try: 

114 params = inspect.signature(fn).parameters 

115 except (TypeError, ValueError): 

116 return False 

117 

118 param = params.get("handler_context") 

119 if param is None: 

120 return False 

121 

122 return param.kind in ( 

123 inspect.Parameter.POSITIONAL_OR_KEYWORD, 

124 inspect.Parameter.KEYWORD_ONLY, 

125 inspect.Parameter.VAR_KEYWORD, 

126 ) 

127 

128 

129class BaseHandler: 

130 def __init__(self, context: Any): 

131 """ 

132 Initialize a new handler to handle a registered type. 

133 

134 :Parameters: 

135 - `context`: reference to pickler/unpickler 

136 

137 """ 

138 self.context = context 

139 

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

141 """ 

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

143 

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

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

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

147 finished. 

148 """ 

149 raise NotImplementedError(f"You must implement flatten() in {self.__class__}") 

150 

151 def restore(self, data: dict[str, Any]) -> Any: 

152 """ 

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

154 representation `obj` and return it. 

155 """ 

156 raise NotImplementedError(f"You must implement restore() in {self.__class__}") 

157 

158 @classmethod 

159 def handles(self, cls: type) -> type: 

160 """ 

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

162 e.g.:: 

163 

164 @MyCustomHandler.handles 

165 class MyCustomClass: 

166 def __reduce__(self): 

167 ... 

168 """ 

169 registry.register(cls, self) 

170 return cls 

171 

172 def __call__(self, context: ContextType) -> "BaseHandler": 

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

174 

175 :Parameters: 

176 - `context`: reference to pickler/unpickler 

177 """ 

178 self.context = context 

179 return self 

180 

181 

182class ArrayHandler(BaseHandler): 

183 """Flatten and restore array.array objects""" 

184 

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

186 data["typecode"] = obj.typecode 

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

188 return data 

189 

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

191 typecode = data["typecode"] 

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

193 if typecode == "c": 

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

195 return array.array(typecode, values) 

196 

197 

198ArrayHandler.handles(array.array) 

199 

200 

201class DatetimeHandler(BaseHandler): 

202 """Custom handler for datetime objects 

203 

204 Datetime objects use __reduce_ex__, and they generate binary strings 

205 encoding the payload. This handler encodes that payload to reconstruct 

206 the object. 

207 

208 """ 

209 

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

211 pickler = self.context 

212 if not pickler.unpicklable: 

213 if hasattr(obj, "isoformat"): 

214 result = obj.isoformat() 

215 else: 

216 result = str(obj) 

217 return result 

218 cls, args = obj.__reduce_ex__(util.PICKLE_PROTOCOL) # type: ignore[str-unpack] 

219 flatten = pickler.flatten 

220 payload = util.b64encode(args[0]) # type: ignore[arg-type] 

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

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

223 return data 

224 

225 def restore(self, data: dict[str, Any]) -> Any: 

226 cls, args = data["__reduce__"] 

227 unpickler = self.context 

228 restore = unpickler.restore 

229 cls = restore(cls, reset=False) 

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

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

232 return cls.__new__(cls, *params) 

233 

234 

235DatetimeHandler.handles(datetime.datetime) 

236DatetimeHandler.handles(datetime.date) 

237DatetimeHandler.handles(datetime.time) 

238 

239 

240class RegexHandler(BaseHandler): 

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

242 

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

244 pattern = obj.pattern 

245 if isinstance(pattern, bytes): 

246 pattern = self.context._flatten_bytestring(pattern) 

247 data["pattern"] = pattern 

248 data["flags"] = obj.flags 

249 return data 

250 

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

252 pattern = data["pattern"] 

253 if isinstance(pattern, dict): 

254 if tags.B85 in pattern: 

255 pattern = self.context._restore_base85(pattern) 

256 else: 

257 pattern = self.context._restore_base64(pattern) 

258 return re.compile(pattern, data.get("flags", 0)) 

259 

260 

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

262 

263 

264class QueueHandler(BaseHandler): 

265 """Opaquely serializes Queue objects 

266 

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

268 Construct a new Queue instance when restoring. 

269 

270 """ 

271 

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

273 return data 

274 

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

276 return queue.Queue() 

277 

278 

279QueueHandler.handles(queue.Queue) 

280 

281 

282class CloneFactory: 

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

284 

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

286 self.exemplar = exemplar 

287 

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

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

290 return clone(self.exemplar) 

291 

292 def __repr__(self) -> str: 

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

294 

295 

296class UUIDHandler(BaseHandler): 

297 """Serialize uuid.UUID objects""" 

298 

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

300 data["hex"] = obj.hex 

301 return data 

302 

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

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

305 

306 

307UUIDHandler.handles(uuid.UUID) 

308 

309 

310class LockHandler(BaseHandler): 

311 """Serialize threading.Lock objects""" 

312 

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

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

315 return data 

316 

317 def restore(self, data: dict[str, Any]) -> Any: 

318 lock = threading.Lock() 

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

320 lock.acquire() 

321 return lock 

322 

323 

324_lock = threading.Lock() 

325LockHandler.handles(_lock.__class__) 

326 

327 

328class TextIOHandler(BaseHandler): 

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

330 

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

332 return None 

333 

334 def restore(self, data: dict[str, Any]) -> NoReturn: 

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

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

337 

338 

339TextIOHandler.handles(io.TextIOWrapper) 

340 

341 

342class PassthroughHandler(BaseHandler): 

343 """ 

344 Hand objects to the backend untouched instead of flattening them. 

345 

346 Backends can support types that jsonpickle would otherwise flatten into a 

347 py/object payload. simplejson in use_decimal mode, for example, writes 

348 decimal.Decimal out as a plain json number. You will normally want to register 

349 this using base=True. 

350 

351 This handler is not registered for any type by default. The backend must know 

352 how to encode and decode the object, so ensure that it can before using this! 

353 

354 Example usage:: 

355 

356 jsonpickle.handlers.register( 

357 decimal.Decimal, PassthroughHandler, base=True 

358 ) 

359 

360 """ 

361 

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

363 # the pickler logged a reference before dispatching here, but the 

364 # backend writes obj as an opaque value that no py/id entry can 

365 # point at. therefore, drop the reference so that a repeat of the 

366 # same instance is encoded again instead of becoming a dangling reference 

367 self.context._unlog_ref(obj) 

368 return obj 

369 

370 def restore(self, data: dict[str, Any]) -> NoReturn: 

371 """ 

372 Restore should never get called because flatten() returns obj 

373 """ 

374 raise AssertionError("Restoring PassthroughHandler is not supported")