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(self, dct: Dict[str, Any], backend: str, obj: ModuleType, name: str):
129 try:
130 dct[backend] = getattr(obj, name)
131 except AttributeError:
132 self.remove_backend(backend)
133 return False
134 return True
135
136 def load_backend(
137 self,
138 name: str,
139 dumps: str = 'dumps',
140 loads: str = 'loads',
141 loads_exc: Union[str, Type[Exception]] = ValueError,
142 ) -> bool:
143 """Load a JSON backend by name.
144
145 This method loads a backend and sets up references to that
146 backend's loads/dumps functions and exception classes.
147
148 :param dumps: is the name of the backend's encode method.
149 The method should take an object and return a string.
150 Defaults to 'dumps'.
151 :param loads: names the backend's method for the reverse
152 operation -- returning a Python object from a string.
153 :param loads_exc: can be either the name of the exception class
154 used to denote decoding errors, or it can be a direct reference
155 to the appropriate exception class itself. If it is a name,
156 then the assumption is that an exception class of that name
157 can be found in the backend module's namespace.
158 :param load: names the backend's 'load' method.
159 :param dump: names the backend's 'dump' method.
160 :rtype bool: True on success, False if the backend could not be loaded.
161
162 """
163 try:
164 # Load the JSON backend
165 mod = __import__(name)
166 except ImportError:
167 return False
168
169 # Handle submodules, e.g. django.utils.simplejson
170 try:
171 for attr in name.split('.')[1:]:
172 mod = getattr(mod, attr)
173 except AttributeError:
174 return False
175
176 if not self._store(self._encoders, name, mod, dumps) or not self._store(
177 self._decoders, name, mod, loads
178 ):
179 return False
180
181 if isinstance(loads_exc, str):
182 # This backend's decoder exception is part of the backend
183 if not self._store(self._decoder_exceptions, name, mod, loads_exc):
184 return False
185 else:
186 # simplejson uses ValueError
187 self._decoder_exceptions[name] = loads_exc
188
189 # Setup the default args and kwargs for this encoder/decoder
190 self._encoder_options.setdefault(name, ([], {})) # type: ignore
191 self._decoder_options.setdefault(name, ([], {}))
192
193 # Add this backend to the list of candidate backends
194 self._backend_names.append(name)
195
196 # Indicate that we successfully loaded a JSON backend
197 self._verified = True
198 return True
199
200 def remove_backend(self, name: str) -> None:
201 """Remove all entries for a particular backend."""
202 self._encoders.pop(name, None)
203 self._decoders.pop(name, None)
204 self._decoder_exceptions.pop(name, None)
205 self._decoder_options.pop(name, None)
206 self._encoder_options.pop(name, None)
207 if name in self._backend_names:
208 self._backend_names.remove(name)
209 self._verified = bool(self._backend_names)
210
211 def backend_encode(
212 self,
213 name: str,
214 obj: Any,
215 indent: Optional[int] = None,
216 separators: Optional[str] = None,
217 ):
218 optargs, optkwargs = self._encoder_options.get(name, ([], {}))
219 encoder_kwargs = optkwargs.copy()
220 if indent is not None:
221 encoder_kwargs['indent'] = indent # type: ignore[assignment]
222 if separators is not None:
223 encoder_kwargs['separators'] = separators # type: ignore[assignment]
224 encoder_args = (obj,) + tuple(optargs)
225 return self._encoders[name](*encoder_args, **encoder_kwargs)
226
227 def backend_decode(self, name: str, string: str) -> Any:
228 optargs, optkwargs = self._decoder_options.get(name, ((), {}))
229 decoder_kwargs = optkwargs.copy()
230 return self._decoders[name](string, *optargs, **decoder_kwargs)
231
232 def set_preferred_backend(self, name: str) -> None:
233 """
234 Set the preferred json backend.
235
236 If a preferred backend is set then jsonpickle tries to use it
237 before any other backend.
238
239 For example::
240
241 set_preferred_backend('simplejson')
242
243 If the backend is not one of the built-in jsonpickle backends
244 (json/simplejson) then you must load the backend
245 prior to calling set_preferred_backend.
246
247 AssertionError is raised if the backend has not been loaded.
248
249 """
250 if name in self._backend_names:
251 self._backend_names.remove(name)
252 self._backend_names.insert(0, name)
253 else:
254 errmsg = 'The "%s" backend has not been loaded.' % name
255 raise AssertionError(errmsg)
256
257 def set_encoder_options(self, name: str, *args: Any, **kwargs: Any) -> None:
258 """
259 Associate encoder-specific options with an encoder.
260
261 After calling set_encoder_options, any calls to jsonpickle's
262 encode method will pass the supplied args and kwargs along to
263 the appropriate backend's encode method.
264
265 For example::
266
267 set_encoder_options('simplejson', sort_keys=True, indent=4)
268
269 See the appropriate encoder's documentation for details about
270 the supported arguments and keyword arguments.
271
272 WARNING: If you pass sort_keys=True, and the object to encode
273 contains ``__slots__``, and you set ``warn`` to True,
274 a TypeError will be raised!
275 """
276 self._encoder_options[name] = (args, kwargs)
277
278 def set_decoder_options(self, name: str, *args: Any, **kwargs: Any) -> None:
279 """
280 Associate decoder-specific options with a decoder.
281
282 After calling set_decoder_options, any calls to jsonpickle's
283 decode method will pass the supplied args and kwargs along to
284 the appropriate backend's decode method.
285
286 For example::
287
288 set_decoder_options('simplejson', encoding='utf8', cls=JSONDecoder)
289
290 See the appropriate decoder's documentation for details about
291 the supported arguments and keyword arguments.
292
293 """
294 self._decoder_options[name] = (args, kwargs)
295
296
297json = JSONBackend()