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

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

109 statements  

1from types import ModuleType 

2from typing import Any, Dict, Optional, Type, Union 

3 

4 

5class JSONBackend: 

6 """Manages encoding and decoding using various backends. 

7 

8 It tries these modules in this order: 

9 simplejson, json, ujson 

10 

11 simplejson is a fast and popular backend and is tried first. 

12 json comes with Python and is tried second. 

13 

14 """ 

15 

16 def _verify(self) -> None: 

17 """Ensures that we've loaded at least one JSON backend.""" 

18 if self._verified: 

19 return 

20 raise AssertionError("jsonpickle could not load any json modules") 

21 

22 def encode( 

23 self, obj: Any, indent: Optional[int] = None, separators: Optional[Any] = None 

24 ) -> str: 

25 """ 

26 Attempt to encode an object into JSON. 

27 

28 This tries the loaded backends in order and passes along the last 

29 exception if no backend is able to encode the object. 

30 

31 """ 

32 self._verify() 

33 

34 if not self._fallthrough: 

35 name = self._backend_names[0] 

36 return self.backend_encode(name, obj, indent=indent, separators=separators) 

37 

38 for idx, name in enumerate(self._backend_names): 

39 try: 

40 return self.backend_encode( 

41 name, obj, indent=indent, separators=separators 

42 ) 

43 except Exception as e: 

44 if idx == len(self._backend_names) - 1: 

45 raise e 

46 

47 # def dumps 

48 dumps = encode 

49 

50 def decode(self, string: str) -> Any: 

51 """ 

52 Attempt to decode an object from a JSON string. 

53 

54 This tries the loaded backends in order and passes along the last 

55 exception if no backends are able to decode the string. 

56 

57 """ 

58 self._verify() 

59 

60 if not self._fallthrough: 

61 name = self._backend_names[0] 

62 return self.backend_decode(name, string) 

63 

64 for idx, name in enumerate(self._backend_names): 

65 try: 

66 return self.backend_decode(name, string) 

67 except self._decoder_exceptions[name] as e: 

68 if idx == len(self._backend_names) - 1: 

69 raise e 

70 else: 

71 pass # and try a more forgiving encoder 

72 

73 # def loads 

74 loads = decode 

75 

76 def __init__(self, fallthrough: bool = True) -> None: 

77 # Whether we should fallthrough to the next backend 

78 self._fallthrough = fallthrough 

79 # The names of backends that have been successfully imported 

80 self._backend_names = [] 

81 

82 # A dictionary mapping backend names to encode/decode functions 

83 self._encoders = {} 

84 self._decoders = {} 

85 

86 # Options to pass to specific encoders 

87 self._encoder_options = {} 

88 

89 # Options to pass to specific decoders 

90 self._decoder_options = {} 

91 

92 # The exception class that is thrown when a decoding error occurs 

93 self._decoder_exceptions = {} 

94 

95 # Whether we've loaded any backends successfully 

96 self._verified = False 

97 

98 self.load_backend("simplejson") 

99 self.load_backend("json") 

100 self.load_backend("ujson") 

101 

102 # Defaults for various encoders 

103 json_opts = ((), {"sort_keys": False}) 

104 self._encoder_options = { 

105 "ujson": ((), {"sort_keys": False, "escape_forward_slashes": False}), 

106 "json": json_opts, 

107 "simplejson": json_opts, 

108 "django.util.simplejson": json_opts, 

109 } 

110 

111 def enable_fallthrough(self, enable: bool) -> None: 

112 """ 

113 Disable jsonpickle's fallthrough-on-error behavior 

114 

115 By default, jsonpickle tries the next backend when decoding or 

116 encoding using a backend fails. 

117 

118 This can make it difficult to force jsonpickle to use a specific 

119 backend, and catch errors, because the error will be suppressed and 

120 may not be raised by the subsequent backend. 

121 

122 Calling `enable_backend(False)` will make jsonpickle immediately 

123 re-raise any exceptions raised by the backends. 

124 

125 """ 

126 self._fallthrough = enable 

127 

128 def _store( 

129 self, dct: Dict[str, Any], backend: str, obj: ModuleType, name: str 

130 ) -> bool: 

131 try: 

132 dct[backend] = getattr(obj, name) 

133 except AttributeError: 

134 self.remove_backend(backend) 

135 return False 

136 return True 

137 

138 def load_backend( 

139 self, 

140 name: str, 

141 dumps: str = "dumps", 

142 loads: str = "loads", 

143 loads_exc: Union[str, Type[Exception]] = ValueError, 

144 ) -> bool: 

145 """Load a JSON backend by name. 

146 

147 This method loads a backend and sets up references to that 

148 backend's loads/dumps functions and exception classes. 

149 

150 :param dumps: is the name of the backend's encode method. 

151 The method should take an object and return a string. 

152 Defaults to 'dumps'. 

153 :param loads: names the backend's method for the reverse 

154 operation -- returning a Python object from a string. 

155 :param loads_exc: can be either the name of the exception class 

156 used to denote decoding errors, or it can be a direct reference 

157 to the appropriate exception class itself. If it is a name, 

158 then the assumption is that an exception class of that name 

159 can be found in the backend module's namespace. 

160 :param load: names the backend's 'load' method. 

161 :param dump: names the backend's 'dump' method. 

162 :rtype bool: True on success, False if the backend could not be loaded. 

163 

164 """ 

165 try: 

166 # Load the JSON backend 

167 mod = __import__(name) 

168 except ImportError: 

169 return False 

170 

171 # Handle submodules, e.g. django.utils.simplejson 

172 try: 

173 for attr in name.split(".")[1:]: 

174 mod = getattr(mod, attr) 

175 except AttributeError: 

176 return False 

177 

178 if not self._store(self._encoders, name, mod, dumps) or not self._store( 

179 self._decoders, name, mod, loads 

180 ): 

181 return False 

182 

183 if isinstance(loads_exc, str): 

184 # This backend's decoder exception is part of the backend 

185 if not self._store(self._decoder_exceptions, name, mod, loads_exc): 

186 return False 

187 else: 

188 # simplejson uses ValueError 

189 self._decoder_exceptions[name] = loads_exc 

190 

191 # Setup the default args and kwargs for this encoder/decoder 

192 self._encoder_options.setdefault(name, ([], {})) # type: ignore 

193 self._decoder_options.setdefault(name, ([], {})) 

194 

195 # Add this backend to the list of candidate backends 

196 self._backend_names.append(name) 

197 

198 # Indicate that we successfully loaded a JSON backend 

199 self._verified = True 

200 return True 

201 

202 def remove_backend(self, name: str) -> None: 

203 """Remove all entries for a particular backend.""" 

204 self._encoders.pop(name, None) 

205 self._decoders.pop(name, None) 

206 self._decoder_exceptions.pop(name, None) 

207 self._decoder_options.pop(name, None) 

208 self._encoder_options.pop(name, None) 

209 if name in self._backend_names: 

210 self._backend_names.remove(name) 

211 self._verified = bool(self._backend_names) 

212 

213 def backend_encode( 

214 self, 

215 name: str, 

216 obj: Any, 

217 indent: Optional[int] = None, 

218 separators: Optional[str] = None, 

219 ) -> str: 

220 optargs, optkwargs = self._encoder_options.get(name, ([], {})) 

221 encoder_kwargs = optkwargs.copy() 

222 if indent is not None: 

223 encoder_kwargs["indent"] = indent # type: ignore[assignment] 

224 if separators is not None: 

225 encoder_kwargs["separators"] = separators # type: ignore[assignment] 

226 encoder_args = (obj,) + tuple(optargs) 

227 return self._encoders[name](*encoder_args, **encoder_kwargs) # type: ignore[no-any-return] 

228 

229 def backend_decode(self, name: str, string: str) -> Any: 

230 optargs, optkwargs = self._decoder_options.get(name, ((), {})) 

231 decoder_kwargs = optkwargs.copy() 

232 return self._decoders[name](string, *optargs, **decoder_kwargs) 

233 

234 def set_preferred_backend(self, name: str) -> None: 

235 """ 

236 Set the preferred json backend. 

237 

238 If a preferred backend is set then jsonpickle tries to use it 

239 before any other backend. 

240 

241 For example:: 

242 

243 set_preferred_backend('simplejson') 

244 

245 If the backend is not one of the built-in jsonpickle backends 

246 (json/simplejson) then you must load the backend 

247 prior to calling set_preferred_backend. 

248 

249 AssertionError is raised if the backend has not been loaded. 

250 

251 """ 

252 if name in self._backend_names: 

253 self._backend_names.remove(name) 

254 self._backend_names.insert(0, name) 

255 else: 

256 errmsg = 'The "%s" backend has not been loaded.' % name 

257 raise AssertionError(errmsg) 

258 

259 def set_encoder_options(self, name: str, *args: Any, **kwargs: Any) -> None: 

260 """ 

261 Associate encoder-specific options with an encoder. 

262 

263 After calling set_encoder_options, any calls to jsonpickle's 

264 encode method will pass the supplied args and kwargs along to 

265 the appropriate backend's encode method. 

266 

267 For example:: 

268 

269 set_encoder_options('simplejson', sort_keys=True, indent=4) 

270 

271 See the appropriate encoder's documentation for details about 

272 the supported arguments and keyword arguments. 

273 

274 WARNING: If you pass sort_keys=True, and the object to encode 

275 contains ``__slots__``, and you set ``warn`` to True, 

276 a TypeError will be raised! 

277 """ 

278 self._encoder_options[name] = (args, kwargs) 

279 

280 def set_decoder_options(self, name: str, *args: Any, **kwargs: Any) -> None: 

281 """ 

282 Associate decoder-specific options with a decoder. 

283 

284 After calling set_decoder_options, any calls to jsonpickle's 

285 decode method will pass the supplied args and kwargs along to 

286 the appropriate backend's decode method. 

287 

288 For example:: 

289 

290 set_decoder_options('simplejson', encoding='utf8', cls=JSONDecoder) 

291 

292 See the appropriate decoder's documentation for details about 

293 the supported arguments and keyword arguments. 

294 

295 """ 

296 self._decoder_options[name] = (args, kwargs) 

297 

298 

299json = JSONBackend()