Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_load_list_multithreaded(self):
dat = datetime.datetime(year=2018, month=7, day=8, hour=21, minute=34,
tzinfo=datetime.timezone.utc)
list_ = [1, 2, 3, [4, 5, [dat]]]
expectation = [1, 2, 3, [4, 5, ['2018-07-08T21:34:00Z']]]
self.assertEqual(list_, jsons.load(expectation, tasks=3,
task_type=Thread))
with self.assertRaises(JsonsError):
jsons.load(expectation, tasks=-1,
task_type=Thread)
self.assertEqual([1], jsons.load(['1'], List[int], tasks=2,
task_type=Thread))
# More tasks than elements should still work.
self.assertEqual([1, 1, 1, 1], jsons.load(['1', '1', '1', '1'],
List[int], tasks=16,
task_type=Thread))
def default_union_deserializer(obj: object, cls: Union, **kwargs) -> object:
"""
Deserialize an object to any matching type of the given union. The first
successful deserialization is returned.
:param obj: The object that needs deserializing.
:param cls: The Union type with a generic (e.g. Union[str, int]).
:param kwargs: Any keyword arguments that are passed through the
deserialization process.
:return: An object of the first type of the Union that could be
deserialized successfully.
"""
for sub_type in get_union_params(cls):
try:
return load(obj, sub_type, **kwargs)
except JsonsError:
pass # Try the next one.
else:
args_msg = ', '.join([get_class_name(cls_)
for cls_ in get_union_params(cls)])
err_msg = ('Could not match the object of type "{}" to any type of '
'the Union: {}'.format(type(obj).__name__, args_msg))
raise DeserializationError(err_msg, obj, cls)
"""
cls_ = None
kwargs_ = {**kwargs}
cls_args = get_args(cls)
if cls_args:
cls_ = cls_args[0]
# Mark the cls as 'inferred' so that later it is known where cls came
# from and the precedence of classes can be determined.
kwargs_['_inferred_cls'] = True
if tasks == 1:
result = [load(elem, cls=cls_, tasks=1, **kwargs_) for elem in obj]
elif tasks > 1:
result = multi_task(load, obj, tasks, task_type, cls_, **kwargs_)
else:
raise JsonsError('Invalid number of tasks: {}'.format(tasks))
return result
def _do_load(json_obj: object,
deserializer: callable,
cls: type,
initial: bool,
**kwargs):
try:
result = deserializer(json_obj, cls, **kwargs)
validate(result, cls, kwargs['fork_inst'])
except Exception as err:
clear()
if isinstance(err, JsonsError):
raise
raise DeserializationError(str(err), json_obj, cls)
else:
if initial:
# Clear all lru caches right before returning the initial call.
clear()
return result
:param argument: the name of the argument in question.
"""
JsonsError.__init__(self, message)
ValueError.__init__(self, message)
self._argument = argument
@property
def argument(self) -> str:
"""
The argument in question.
:return: the name of the argument.
"""
return self._argument
class DeserializationError(JsonsError):
"""
Raised when deserialization failed for some reason.
"""
def __init__(self, message: str, source: object, target: Optional[type]):
"""
Constructor.
:param message: the message describing the problem.
:param source: the object that was to be deserialized.
:param target: the type to which `source` was to be deserialized.
"""
JsonsError.__init__(self, message)
self._source = source
self._target = target
@property
def source(self) -> object:
"""
The object that was to be deserialized.
:return: the object that was to be deserialized.
"""
return self._source
@property
def target(self) -> Optional[type]:
"""
The target type to which `source` was to be deserialized.
:return: the type to which `source` was to be deserialized.
"""
return self._target
class SerializationError(JsonsError):
"""
Raised when serialization failed for some reason.
"""
class RecursionDetectedError(SerializationError):
"""
Raised when a recursive structure was detected and a stack overflow was
prevented during serialization.
"""
class DecodeError(DeserializationError, JSONDecodeError):
"""
Raised when decoding a string or bytes to Python types failed. This error
is actually a wrapper around `json.JSONDecodeError`.
cls = get_cls_from_str(cls, json_obj, fork_inst)
cls, meta_hints = _check_and_get_cls_and_meta_hints(
json_obj, cls, fork_inst, kwargs.get('_inferred_cls', False))
deserializer = _get_deserializer(cls, fork_inst)
kwargs_ = {
'strict': strict,
'fork_inst': fork_inst,
'attr_getters': attr_getters,
'meta_hints': meta_hints,
**kwargs
}
try:
return deserializer(json_obj, cls, **kwargs_)
except Exception as err:
if isinstance(err, JsonsError):
raise
raise DeserializationError(str(err), json_obj, cls)
:param target_name: the name of the type that was the target type.
"""
DeserializationError.__init__(self, message, source, None)
self._target_name = target_name
@property
def target_name(self) -> str:
"""
The name of the type that was unsuccessfully attempted to deserialize
into.
:return: the name of the type that was to be the target type.
"""
return self._target_name
class InvalidDecorationError(JsonsError):
"""
Raised when a jsons decorator was wrongly used.
"""
def __init__(self, message: str):
"""
Constructor.
:param message: the message of this error.
"""
JsonsError.__init__(self, message)
Base class for all `jsons` errors.
"""
def __init__(self, message: str):
"""
Constructor.
:param message: the message describing the problem.
"""
Exception.__init__(self, message)
self._message = message
@property
def message(self):
return self._message
class ValidationError(JsonsError):
"""
Raised when the validation of an object failed.
"""
class ArgumentError(JsonsError, ValueError):
"""
Raised when serialization or deserialization went wrong caused by a wrong
argument when serializing or deserializing.
"""
def __init__(self, message: str, argument: str):
"""
Constructor.
:param message: the message describing the problem.
:param argument: the name of the argument in question.
"""
def default_union_deserializer(obj: object, cls: Union, **kwargs) -> object:
"""
Deserialize an object to any matching type of the given union. The first
successful deserialization is returned.
:param obj: The object that needs deserializing.
:param cls: The Union type with a generic (e.g. Union[str, int]).
:param kwargs: Any keyword arguments that are passed through the
deserialization process.
:return: An object of the first type of the Union that could be
deserialized successfully.
"""
for sub_type in cls.__args__:
try:
return _main_impl.load(obj, sub_type, **kwargs)
except JsonsError:
pass # Try the next one.
else:
args_msg = ', '.join([get_class_name(cls_) for cls_ in cls.__args__])
err_msg = ('Could not match the object of type "{}" to any type of '
'the Union: {}'.format(str(cls), args_msg))
raise DeserializationError(err_msg, obj, cls)