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
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
1from types import ModuleType
2from typing import Any, Dict, Optional, Type, Union
5class JSONBackend:
6 """Manages encoding and decoding using various backends.
8 It tries these modules in this order:
9 simplejson, json, ujson
11 simplejson is a fast and popular backend and is tried first.
12 json comes with Python and is tried second.
14 """
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")
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.
28 This tries the loaded backends in order and passes along the last
29 exception if no backend is able to encode the object.
31 """
32 self._verify()
34 if not self._fallthrough:
35 name = self._backend_names[0]
36 return self.backend_encode(name, obj, indent=indent, separators=separators)
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
47 # def dumps
48 dumps = encode
50 def decode(self, string: str) -> Any:
51 """
52 Attempt to decode an object from a JSON string.
54 This tries the loaded backends in order and passes along the last
55 exception if no backends are able to decode the string.
57 """
58 self._verify()
60 if not self._fallthrough:
61 name = self._backend_names[0]
62 return self.backend_decode(name, string)
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
73 # def loads
74 loads = decode
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 = []
82 # A dictionary mapping backend names to encode/decode functions
83 self._encoders = {}
84 self._decoders = {}
86 # Options to pass to specific encoders
87 self._encoder_options = {}
89 # Options to pass to specific decoders
90 self._decoder_options = {}
92 # The exception class that is thrown when a decoding error occurs
93 self._decoder_exceptions = {}
95 # Whether we've loaded any backends successfully
96 self._verified = False
98 self.load_backend("simplejson")
99 self.load_backend("json")
100 self.load_backend("ujson")
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 }
111 def enable_fallthrough(self, enable: bool) -> None:
112 """
113 Disable jsonpickle's fallthrough-on-error behavior
115 By default, jsonpickle tries the next backend when decoding or
116 encoding using a backend fails.
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.
122 Calling `enable_backend(False)` will make jsonpickle immediately
123 re-raise any exceptions raised by the backends.
125 """
126 self._fallthrough = enable
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
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.
147 This method loads a backend and sets up references to that
148 backend's loads/dumps functions and exception classes.
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.
164 """
165 try:
166 # Load the JSON backend
167 mod = __import__(name)
168 except ImportError:
169 return False
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
178 if not self._store(self._encoders, name, mod, dumps) or not self._store(
179 self._decoders, name, mod, loads
180 ):
181 return False
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
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, ([], {}))
195 # Add this backend to the list of candidate backends
196 self._backend_names.append(name)
198 # Indicate that we successfully loaded a JSON backend
199 self._verified = True
200 return True
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)
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]
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)
234 def set_preferred_backend(self, name: str) -> None:
235 """
236 Set the preferred json backend.
238 If a preferred backend is set then jsonpickle tries to use it
239 before any other backend.
241 For example::
243 set_preferred_backend('simplejson')
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.
249 AssertionError is raised if the backend has not been loaded.
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)
259 def set_encoder_options(self, name: str, *args: Any, **kwargs: Any) -> None:
260 """
261 Associate encoder-specific options with an encoder.
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.
267 For example::
269 set_encoder_options('simplejson', sort_keys=True, indent=4)
271 See the appropriate encoder's documentation for details about
272 the supported arguments and keyword arguments.
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)
280 def set_decoder_options(self, name: str, *args: Any, **kwargs: Any) -> None:
281 """
282 Associate decoder-specific options with a decoder.
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.
288 For example::
290 set_decoder_options('simplejson', encoding='utf8', cls=JSONDecoder)
292 See the appropriate decoder's documentation for details about
293 the supported arguments and keyword arguments.
295 """
296 self._decoder_options[name] = (args, kwargs)
299json = JSONBackend()