Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Deserialize a dict by deserializing all instances of that dict.
:param obj: the dict that needs deserializing.
:param key_transformer: a function that transforms the keys to a different
style (e.g. PascalCase).
:param cls: not used.
:param kwargs: any keyword arguments.
:return: a deserialized dict instance.
"""
key_tfr = key_transformer or (lambda key: key)
cls_args = get_args(cls)
kwargs_ = {**kwargs, 'key_transformer': key_transformer}
if len(cls_args) == 2:
cls_k, cls_v = cls_args
kwargs_k = {**kwargs_, 'cls': cls_k}
kwargs_v = {**kwargs_, 'cls': cls_v}
res = {load(key_tfr(k), **kwargs_k): load(obj[k], **kwargs_v)
for k in obj}
else:
res = {key_tfr(key): load(obj[key], **kwargs_)
for key in obj}
return res
cls_key = '/{}'.format(sig_key)
cls_str_from_meta = meta_hints.get(cls_key, None)
new_hints = meta_hints
cls_from_meta = None
if cls_str_from_meta:
cls_from_meta = get_cls_from_str(
cls_str_from_meta, obj, kwargs['fork_inst'])
# Rebuild the class hints: cls_key becomes the new root.
new_hints = {
key.replace(cls_key, '/'): meta_hints[key]
for key in meta_hints
if key != '/'
}
cls_ = determine_precedence(cls=cls, cls_from_meta=cls_from_meta,
cls_from_type=None, inferred_cls=True)
value = load(obj[sig_key], cls_, meta_hints=new_hints, **kwargs)
return value
:param obj: the tuple that needs deserializing.
:param cls: the type optionally with a generic (e.g. Tuple[str, int]).
:param kwargs: any keyword arguments.
:return: a deserialized tuple instance.
"""
if hasattr(cls, '_fields'):
return default_namedtuple_deserializer(obj, cls, **kwargs)
cls_args = get_args(cls)
if cls_args:
tuple_types = getattr(cls, '__tuple_params__', cls_args)
if tuple_with_ellipsis(cls):
tuple_types = [tuple_types[0]] * len(obj)
list_ = [load(value, tuple_types[i], **kwargs)
for i, value in enumerate(obj)]
else:
list_ = [load(value, **kwargs) for i, value in enumerate(obj)]
return tuple(list_)
def load(cls: type, json_obj: object, **kwargs) -> object:
"""
See ``jsons.load``.
:param kwargs: the keyword args are passed on to the serializer
function.
:param json_obj: the object that is loaded into an instance of `cls`.
:return: this instance in a JSON representation (dict).
"""
return load(json_obj, cls, fork_inst=cls, **kwargs)
# _field_types has been deprecated in favor of __annotations__ in Python 3.8
if hasattr(cls, '__annotations__'):
field_types = getattr(cls, '__annotations__', {})
else:
field_types = getattr(cls, '_field_types', {})
if field is None:
hint = field_types.get(field_name)
if NoneType not in (get_union_params(hint) or []):
# The value 'None' is not permitted here.
msg = ('No value present in {} for argument "{}"'
.format(obj, field_name))
raise UnfulfilledArgumentError(msg, field_name, obj, cls)
cls_ = field_types.get(field_name) if field_types else None
loaded_field = load(field, cls_, **kwargs)
args.append(loaded_field)
inst = cls(*args)
return inst
def default_string_deserializer(obj: str,
cls: Optional[type] = None,
**kwargs) -> object:
"""
Deserialize a string. If the given ``obj`` can be parsed to a date, a
``datetime`` instance is returned.
:param obj: the string that is to be deserialized.
:param cls: not used.
:param kwargs: any keyword arguments.
:return: the deserialized obj.
"""
try:
result = load(obj, datetime, **kwargs)
except DeserializationError:
result = default_primitive_deserializer(obj, str)
return result
:param tasks: the allowed number of tasks (threads or processes).
:param task_type: the type that is used for multitasking.
:param kwargs: any keyword arguments.
:return: a deserialized list instance.
"""
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