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

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 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[str], data: dict[str, Any]) -> HandlerReturn: 

244 data["pattern"] = obj.pattern 

245 data["flags"] = obj.flags 

246 return data 

247 

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

249 return re.compile(data["pattern"], data.get("flags", 0)) 

250 

251 

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

253 

254 

255class QueueHandler(BaseHandler): 

256 """Opaquely serializes Queue objects 

257 

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

259 Construct a new Queue instance when restoring. 

260 

261 """ 

262 

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

264 return data 

265 

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

267 return queue.Queue() 

268 

269 

270QueueHandler.handles(queue.Queue) 

271 

272 

273class CloneFactory: 

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

275 

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

277 self.exemplar = exemplar 

278 

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

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

281 return clone(self.exemplar) 

282 

283 def __repr__(self) -> str: 

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

285 

286 

287class UUIDHandler(BaseHandler): 

288 """Serialize uuid.UUID objects""" 

289 

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

291 data["hex"] = obj.hex 

292 return data 

293 

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

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

296 

297 

298UUIDHandler.handles(uuid.UUID) 

299 

300 

301class LockHandler(BaseHandler): 

302 """Serialize threading.Lock objects""" 

303 

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

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

306 return data 

307 

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

309 lock = threading.Lock() 

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

311 lock.acquire() 

312 return lock 

313 

314 

315_lock = threading.Lock() 

316LockHandler.handles(_lock.__class__) 

317 

318 

319class TextIOHandler(BaseHandler): 

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

321 

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

323 return None 

324 

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

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

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

328 

329 

330TextIOHandler.handles(io.TextIOWrapper) 

331 

332 

333class PassthroughHandler(BaseHandler): 

334 """ 

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

336 

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

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

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

340 this using base=True. 

341 

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

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

344 

345 Example usage:: 

346 

347 jsonpickle.handlers.register( 

348 decimal.Decimal, PassthroughHandler, base=True 

349 ) 

350 

351 """ 

352 

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

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

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

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

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

358 self.context._unlog_ref(obj) 

359 return obj 

360 

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

362 """ 

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

364 """ 

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