comment here

This commit is contained in:
ton
2023-03-18 20:03:34 +07:00
commit 4553a0a589
14513 changed files with 2685043 additions and 0 deletions

View File

@@ -0,0 +1,321 @@
"""
Works with Python3.5+
JSON (de)serialization (jsons) from and to dicts and plain old Python objects.
Works with dataclasses (Python3.7+).
**Example:**
>>> from dataclasses import dataclass
>>> @dataclass
... class Car:
... color: str
>>> @dataclass
... class Person:
... car: Car
... name: str
>>> c = Car('Red')
>>> p = Person(c, 'John')
>>> dumped = dump(p)
>>> dumped['name']
'John'
>>> dumped['car']['color']
'Red'
>>> p_reloaded = load(dumped, Person)
>>> p_reloaded.name
'John'
>>> p_reloaded.car.color
'Red'
Deserialization will work with older Python classes (Python3.5+) given that
type hints are present for custom types (i.e. any type that is not set at
the bottom of this module). Serialization will work with no type hints at
all.
**Example**
>>> class Car:
... def __init__(self, color):
... self.color = color
>>> class Person:
... def __init__(self, car: Car, name):
... self.car = car
... self.name = name
>>> c = Car('Red')
>>> p = Person(c, 'John')
>>> dumped = dump(p)
>>> dumped['name']
'John'
>>> dumped['car']['color']
'Red'
>>> p_reloaded = load(dumped, Person)
>>> p_reloaded.name
'John'
>>> p_reloaded.car.color
'Red'
Alternatively, you can make use of the `JsonSerializable` class.
**Example**
>>> class Car(JsonSerializable):
... def __init__(self, color):
... self.color = color
>>> class Person(JsonSerializable):
... def __init__(self, car: Car, name):
... self.car = car
... self.name = name
>>> c = Car('Red')
>>> p = Person(c, 'John')
>>> dumped = p.json
>>> dumped['name']
'John'
>>> dumped['car']['color']
'Red'
>>> p_reloaded = Person.from_json(dumped)
>>> p_reloaded.name
'John'
>>> p_reloaded.car.color
'Red'
"""
from collections.abc import Mapping
from datetime import datetime, date, time, timezone, timedelta
from decimal import Decimal
from enum import Enum, IntEnum
from pathlib import PurePath
from typing import Union, List, Tuple, Iterable, Optional, DefaultDict, Dict
from uuid import UUID
from jsons._common_impl import NoneType
from jsons._dump_impl import (
dump,
dumps,
dumpb,
)
from jsons._extra_impl import (
announce_class,
suppress_warnings,
suppress_warning,
)
from jsons._fork_impl import fork
from jsons._key_transformers import (
camelcase,
snakecase,
pascalcase,
lispcase,
)
from jsons._lizers_impl import (
get_serializer,
get_deserializer,
set_serializer,
set_deserializer,
)
from jsons._load_impl import (
load,
loads,
loadb,
)
from jsons._package_info import __version__
from jsons._transform_impl import transform
from jsons._validation import (
validate,
get_validator,
set_validator,
)
from jsons.classes.json_serializable import JsonSerializable
from jsons.classes.verbosity import Verbosity
from jsons.deserializers.default_complex import default_complex_deserializer
from jsons.deserializers.default_date import default_date_deserializer
from jsons.deserializers.default_datetime import default_datetime_deserializer
from jsons.deserializers.default_decimal import default_decimal_deserializer
from jsons.deserializers.default_defaultdict import default_defaultdict_deserializer
from jsons.deserializers.default_dict import default_dict_deserializer
from jsons.deserializers.default_enum import default_enum_deserializer
from jsons.deserializers.default_iterable import default_iterable_deserializer
from jsons.deserializers.default_list import default_list_deserializer
from jsons.deserializers.default_mapping import default_mapping_deserializer
from jsons.deserializers.default_nonetype import default_nonetype_deserializer
from jsons.deserializers.default_object import default_object_deserializer
from jsons.deserializers.default_path import default_path_deserializer
from jsons.deserializers.default_primitive import default_primitive_deserializer
from jsons.deserializers.default_string import default_string_deserializer
from jsons.deserializers.default_time import default_time_deserializer
from jsons.deserializers.default_timedelta import default_timedelta_deserializer
from jsons.deserializers.default_timezone import default_timezone_deserializer
from jsons.deserializers.default_tuple import default_tuple_deserializer
from jsons.deserializers.default_union import default_union_deserializer
from jsons.deserializers.default_uuid import default_uuid_deserializer
from jsons.deserializers.default_zone_info import default_zone_info_deserializer
from jsons.exceptions import (
JsonsError,
ValidationError,
SerializationError,
DeserializationError,
DecodeError,
UnfulfilledArgumentError,
InvalidDecorationError
)
from jsons.serializers.default_complex import default_complex_serializer
from jsons.serializers.default_date import default_date_serializer
from jsons.serializers.default_datetime import default_datetime_serializer
from jsons.serializers.default_decimal import default_decimal_serializer
from jsons.serializers.default_dict import default_dict_serializer
from jsons.serializers.default_enum import default_enum_serializer
from jsons.serializers.default_iterable import default_iterable_serializer
from jsons.serializers.default_list import default_list_serializer
from jsons.serializers.default_object import default_object_serializer
from jsons.serializers.default_path import default_path_serializer
from jsons.serializers.default_primitive import default_primitive_serializer
from jsons.serializers.default_time import default_time_serializer
from jsons.serializers.default_timedelta import default_timedelta_serializer
from jsons.serializers.default_timezone import default_timezone_serializer
from jsons.serializers.default_tuple import default_tuple_serializer
from jsons.serializers.default_union import default_union_serializer
from jsons.serializers.default_uuid import default_uuid_serializer
from jsons.serializers.default_zone_info import default_zone_info_serializer
KEY_TRANSFORMER_SNAKECASE = snakecase
KEY_TRANSFORMER_CAMELCASE = camelcase
KEY_TRANSFORMER_PASCALCASE = pascalcase
KEY_TRANSFORMER_LISPCASE = lispcase
__all__ = [
# Functions:
'__version__',
dump.__name__,
dumps.__name__,
dumpb.__name__,
load.__name__,
loads.__name__,
loadb.__name__,
transform.__name__,
fork.__name__,
set_serializer.__name__,
'get_serializer',
set_deserializer.__name__,
'get_deserializer',
'get_validator',
set_validator.__name__,
validate.__name__,
'announce_class',
suppress_warnings.__name__,
suppress_warning.__name__,
# Types:
JsonSerializable.__name__,
Verbosity.__name__,
# Key transformers:
snakecase.__name__,
camelcase.__name__,
pascalcase.__name__,
lispcase.__name__,
'KEY_TRANSFORMER_SNAKECASE',
'KEY_TRANSFORMER_CAMELCASE',
'KEY_TRANSFORMER_PASCALCASE',
'KEY_TRANSFORMER_LISPCASE',
# Errors:
JsonsError.__name__,
ValidationError.__name__,
SerializationError.__name__,
DeserializationError.__name__,
DecodeError.__name__,
UnfulfilledArgumentError.__name__,
InvalidDecorationError.__name__,
# Serializers:
default_tuple_serializer.__name__,
default_dict_serializer.__name__,
default_iterable_serializer.__name__,
default_list_serializer.__name__,
default_enum_serializer.__name__,
default_complex_serializer.__name__,
default_datetime_serializer.__name__,
default_date_serializer.__name__,
default_time_serializer.__name__,
default_timezone_serializer.__name__,
default_timedelta_serializer.__name__,
default_primitive_serializer.__name__,
default_object_serializer.__name__,
default_decimal_serializer.__name__,
default_uuid_serializer.__name__,
default_union_serializer.__name__,
default_path_serializer.__name__,
# Deserializers:
default_list_deserializer.__name__,
default_tuple_deserializer.__name__,
default_union_deserializer.__name__,
default_dict_deserializer.__name__,
default_defaultdict_deserializer.__name__,
default_enum_deserializer.__name__,
default_complex_deserializer.__name__,
default_datetime_deserializer.__name__,
default_date_deserializer.__name__,
default_time_deserializer.__name__,
default_timezone_deserializer.__name__,
default_timedelta_deserializer.__name__,
default_string_deserializer.__name__,
default_nonetype_deserializer.__name__,
default_primitive_deserializer.__name__,
default_mapping_deserializer.__name__,
default_iterable_deserializer.__name__,
default_object_deserializer.__name__,
default_uuid_deserializer.__name__,
default_decimal_deserializer.__name__,
default_path_deserializer.__name__,
]
set_serializer(default_tuple_serializer, (tuple, Tuple))
set_serializer(default_complex_serializer, complex)
set_serializer(default_datetime_serializer, datetime)
set_serializer(default_date_serializer, date)
set_serializer(default_time_serializer, time)
set_serializer(default_timezone_serializer, timezone)
set_serializer(default_timedelta_serializer, timedelta)
set_serializer(default_primitive_serializer, (str, int, float, bool, None))
set_serializer(default_enum_serializer, (Enum, IntEnum)) # must be after primitive_serializer
set_serializer(default_dict_serializer, Mapping, False)
set_serializer(default_list_serializer, (list, List))
set_serializer(default_iterable_serializer, Iterable, False)
set_serializer(default_object_serializer, object, False)
set_serializer(default_uuid_serializer, UUID)
set_serializer(default_decimal_serializer, Decimal)
set_serializer(default_union_serializer, (Union, Optional))
set_serializer(default_path_serializer, PurePath)
set_deserializer(default_list_deserializer, (list, List))
set_deserializer(default_tuple_deserializer, (tuple, Tuple))
set_deserializer(default_union_deserializer, (Union, Optional))
set_deserializer(default_defaultdict_deserializer, DefaultDict)
set_deserializer(default_datetime_deserializer, datetime)
set_deserializer(default_date_deserializer, date)
set_deserializer(default_time_deserializer, time)
set_deserializer(default_timezone_deserializer, timezone)
set_deserializer(default_timedelta_deserializer, timedelta)
set_deserializer(default_string_deserializer, str)
set_deserializer(default_nonetype_deserializer, NoneType)
set_deserializer(default_primitive_deserializer, (int, float, bool))
set_deserializer(default_enum_deserializer, (Enum, IntEnum)) # must be after primitive_deserializer
set_deserializer(default_mapping_deserializer, (Mapping, dict, Dict), False)
set_deserializer(default_iterable_deserializer, Iterable, False)
set_deserializer(default_object_deserializer, object, False)
set_deserializer(default_uuid_deserializer, UUID)
set_deserializer(default_complex_deserializer, complex)
set_deserializer(default_decimal_deserializer, Decimal)
set_deserializer(default_path_deserializer, PurePath)
if default_zone_info_serializer and default_zone_info_deserializer:
from zoneinfo import ZoneInfo
__all__.append(default_zone_info_serializer)
__all__.append(default_zone_info_deserializer)
set_serializer(default_zone_info_serializer, ZoneInfo)
set_deserializer(default_zone_info_deserializer, ZoneInfo)

View File

@@ -0,0 +1,46 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functionality for caching functions.
"""
from collections import deque
from functools import lru_cache, update_wrapper
from typing import Callable
class _Wrapper:
"""
A wrapper around a function that needs to be cached. This wrapper allows
for a single point from which cache can be cleared.
"""
instances = deque([])
def __init__(self, wrapped):
self.wrapped = wrapped
self.instances.append(self)
@lru_cache(typed=True)
def __call__(self, *args, **kwargs):
return self.wrapped(*args, **kwargs)
def cached(decorated: Callable):
"""
Alternative for ``functools.lru_cache``. By decorating a function with
``cached``, you can clear the cache of that function by calling
``clear()``.
:param decorated: the decorated function.
:return: a wrapped function.
"""
wrapper = _Wrapper(decorated)
update_wrapper(wrapper=wrapper, wrapped=decorated)
return wrapper
def clear():
"""
Clear all cache of functions that were cached using ``cached``.
:return: None.
"""
for w in _Wrapper.instances:
w.__call__.cache_clear()

View File

@@ -0,0 +1,170 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains implementations of common functionality that can be used
throughout `jsons`.
"""
import builtins
import warnings
from importlib import import_module
from typing import Callable, Optional, Tuple, TypeVar, Any
from jsons._cache import cached
from jsons._compatibility_impl import get_union_params
from jsons.exceptions import UnknownClassError
NoneType = type(None)
JSON_KEYS = (str, int, float, bool, NoneType)
VALID_TYPES = (str, int, float, bool, list, tuple, set, dict, NoneType)
META_ATTR = '-meta' # The name of the attribute holding meta info.
T = TypeVar('T')
class StateHolder:
"""
This class holds the registered serializers and deserializers.
"""
_fork_counter = 0
_classes_serializers = list()
_classes_deserializers = list()
_serializers = dict()
_deserializers = dict()
_validators = dict()
_classes_validators = list()
_announced_classes = dict()
_suppress_warnings = False
_suppressed_warnings = set()
@classmethod
def _warn(cls, msg, code, *args, **kwargs):
if not cls._suppress_warnings and code not in cls._suppressed_warnings:
msg_ = ('{} Use suppress_warning({}) or suppress_warnings(True) to '
'turn off this message.'.format(msg, code))
warnings.warn(msg_, *args, **kwargs)
@cached
def get_class_name(cls: type,
transformer: Optional[Callable[[str], str]] = None,
fully_qualified: bool = False) -> Optional[str]:
"""
Return the name of a class.
:param cls: the class of which the name if to be returned.
:param transformer: any string transformer, e.g. ``str.lower``.
:param fully_qualified: if ``True`` return the fully qualified name (i.e.
complete with module name).
:return: the name of ``cls``, transformed if a transformer is given.
"""
transformer = transformer or (lambda x: x)
cls_name = _get_special_cases(cls)
if cls_name:
return transformer(cls_name)
cls_name = _get_simple_name(cls)
if fully_qualified:
module = _get_module(cls)
if module:
cls_name = '{}.{}'.format(module, cls_name)
cls_name = transformer(cls_name)
return cls_name
def _get_special_cases(cls: type):
if (hasattr(cls, '__qualname__')
and cls.__qualname__ == 'NewType.<locals>.new_type'):
return cls.__name__
def get_cls_from_str(cls_str: str, source: object, fork_inst) -> type:
cls = getattr(builtins, cls_str, None)
if cls:
return cls
if '[' in cls_str and ']' in cls_str:
return _get_generic_cls_from_str(cls_str, source, fork_inst)
try:
splitted = cls_str.split('.')
module_name = '.'.join(splitted[:-1])
cls_name = splitted[-1]
cls_module = import_module(module_name)
cls = getattr(cls_module, cls_name)
except (ImportError, AttributeError, ValueError):
cls = _lookup_announced_class(cls_str, source, fork_inst)
return cls
def _get_generic_cls_from_str(cls_str: str, source: object, fork_inst) -> type:
# If cls_str represents a generic type, try to parse the sub types.
origin_str, subtypes_str = cls_str.split('[')
subtypes_str = subtypes_str[0:-1] # Remove the ']'.
origin = get_cls_from_str(origin_str, source, fork_inst)
subtypes = [get_cls_from_str(s.strip(), source, fork_inst)
for s in subtypes_str.split(',')]
return origin[tuple(subtypes)]
def determine_precedence(
cls: type,
cls_from_meta: type,
cls_from_type: type,
inferred_cls: bool):
order = [cls, cls_from_meta, cls_from_type]
if inferred_cls:
# The type from a verbose dumped object takes precedence over an
# inferred type (e.g. T in List[T]).
order = [cls_from_meta, cls, cls_from_type]
# Now to return the first element in the order that holds a value.
for elem in order:
if elem:
return elem
def get_cls_and_meta(
json_obj: object,
fork_inst: type) -> Tuple[Optional[type], Optional[dict]]:
if isinstance(json_obj, dict) and META_ATTR in json_obj:
cls_str = json_obj[META_ATTR]['classes']['/']
cls = get_cls_from_str(cls_str, json_obj, fork_inst)
return cls, json_obj[META_ATTR]
return None, None
def can_match_with_none(cls: type):
# Return True if cls allows None; None is a valid value with the given cls.
result = cls in (Any, object, None, NoneType)
if not result:
cls_name = get_class_name(cls).lower()
result = (('union' in cls_name or 'optional' in cls_name)
and NoneType in get_union_params(cls))
return result
def _lookup_announced_class(
cls_str: str,
source: object,
fork_inst: type) -> type:
cls = fork_inst._announced_classes.get(cls_str)
if not cls:
msg = ('Could not find a suitable type for "{}". Make sure it can be '
'imported or that is has been announced.'.format(cls_str))
raise UnknownClassError(msg, source, cls_str)
return cls
def _get_simple_name(cls: type) -> str:
if cls is None:
cls = type(cls)
cls_name = getattr(cls, '__name__', None)
if not cls_name:
cls_name = getattr(cls, '_name', None)
if not cls_name:
cls_name = repr(cls)
cls_name = cls_name.split('[')[0] # Remove generic types.
cls_name = cls_name.split('.')[-1] # Remove any . caused by repr.
cls_name = cls_name.split(r"'>")[0] # Remove any '>.
return cls_name
def _get_module(cls: type) -> Optional[str]:
builtin_module = str.__class__.__module__
module = getattr(cls, '__module__', None)
if module and module != builtin_module:
return module

View File

@@ -0,0 +1,91 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functionality for supporting the compatibility of jsons
with multiple Python versions.
"""
import sys
import typing
from enum import Enum
from jsons._cache import cached
class Flag(Enum):
"""
This is a light version of the Flag enum type that was introduced in
Python3.6. It supports the use of pipes for members (Flag.A | Flag.B).
"""
@classmethod
def _get_inst(cls, value):
try:
result = cls(value)
except ValueError:
pseudo_member = object.__new__(cls)
pseudo_member._value_ = value
contained = [elem.name for elem in cls if elem in pseudo_member]
pseudo_member._name_ = '|'.join(contained)
result = pseudo_member
return result
def __or__(self, other: 'Flag') -> 'Flag':
new_value = other.value | self.value
return self._get_inst(new_value)
def __contains__(self, item: 'Flag') -> bool:
return item.value == self.value & item.value
__ror__ = __or__
@cached
def tuple_with_ellipsis(tup: type) -> bool:
# Python3.5: Tuples have __tuple_use_ellipsis__
# Python3.7: Tuples have __args__
use_el = getattr(tup, '__tuple_use_ellipsis__', None)
if use_el is None:
use_el = tup.__args__[-1] is ...
return use_el
@cached
def get_union_params(un: type) -> list:
# Python3.5: Unions have __union_params__
# Python3.7: Unions have __args__
return getattr(un, '__union_params__', getattr(un, '__args__', None))
@cached
def get_naked_class(cls: type) -> type:
# Python3.5: typing classes have __extra__
# Python3.6: typing classes have __extra__
# Python3.7: typing classes have __origin__
# Return the non-generic class (e.g. dict) of a generic type (e.g. Dict).
return getattr(cls, '__extra__', getattr(cls, '__origin__', cls))
@cached
def get_type_hints(callable_: callable, fallback_ns=None):
# Python3.5: get_type_hints raises on classes without explicit constructor.
# Python3.10: get_type_hints on classes does not take the constructor.
try:
result = typing.get_type_hints(callable_)
except AttributeError:
result = {}
except NameError:
# attempt to resolve in global namespace - this works around an
# issue in 3.7 whereby __init__ created by dataclasses fails
# to find it's context. See https://bugs.python.org/issue34776
if fallback_ns is not None:
context_dict = sys.modules[fallback_ns].__dict__
result = typing.get_type_hints(callable_, globalns=context_dict)
if sys.version_info.minor >= 10 and type(callable_) is type:
annotations_from_init = typing.get_type_hints(callable_.__init__)
if 'return' in annotations_from_init:
# Python3.10: 'return' is a key that holds the returning type.
del annotations_from_init['return']
result = {**result, **annotations_from_init}
return result

View File

@@ -0,0 +1,140 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functionality for ``datetime`` related stuff.
"""
from datetime import datetime, timezone, timedelta, time, date
from typing import Union
RFC3339_DATE_PATTERN = '%Y-%m-%d'
RFC3339_TIME_PATTERN = '%H:%M:%S'
RFC3339_DATETIME_PATTERN = '{}T{}'.format(
RFC3339_DATE_PATTERN, RFC3339_TIME_PATTERN)
def to_str(
dt: Union[datetime, date],
strip_microseconds: bool,
fork_inst: type,
pattern: str = RFC3339_DATETIME_PATTERN) -> str:
offset = get_offset_str(dt, fork_inst)
if not strip_microseconds and getattr(dt, 'microsecond', None):
pattern += '.%f'
return dt.strftime("{}{}".format(pattern, offset))
def get_offset_str(
obj: Union[datetime, date, timedelta],
fork_inst: type) -> str:
"""
Return the textual offset of the given ``obj``.
:param obj: a datetime or timedelta instance.
:param fork_inst: the state holder that is used.
:return: the offset following RFC3339.
"""
result = ''
if isinstance(obj, datetime):
result = _datetime_offset_str(obj, fork_inst)
elif isinstance(obj, timedelta):
result = _timedelta_offset_str(obj)
return result
def get_datetime_inst(obj: str, pattern: str) -> datetime:
"""
Return a datetime instance with timezone info from the given ``obj``.
:param obj: the ``obj`` in RFC3339 format.
:param pattern: the datetime pattern.
:return: a datetime instance with timezone info.
"""
if obj[-1] == 'Z':
result = _datetime_utc_inst(obj, pattern)
elif 'T' in pattern:
result = _datetime_offset_inst(obj, pattern)
else:
result = datetime.strptime(obj, pattern)
return result
def _datetime_offset_str(obj: datetime, fork_inst: type) -> str:
"""
Return a textual offset (e.g. +01:00 or Z) for the given datetime.
:param obj: the datetime instance.
:return: the offset for ``obj``.
"""
tzone = obj.tzinfo
if not tzone:
# datetimes without tzinfo are treated as local times.
fork_inst._warn('The use of datetimes without timezone is dangerous '
'and can lead to undesired results.',
'datetime-without-tz')
tzone = datetime.now(timezone.utc).astimezone().tzinfo
if tzone is timezone.utc or tzone.utc is timezone.utc:
return '+00:00'
offset = 'Z'
if tzone.tzname(None) not in ('UTC', 'UTC+00:00'):
tdelta = tzone.utcoffset(None) or \
getattr(tzone, 'adjusted_offset', tzone.utcoffset(obj))
offset = _timedelta_offset_str(tdelta)
return offset
def _timedelta_offset_str(tdelta: timedelta) -> str:
"""
Return a textual offset (e.g. +01:00 or Z) for the given timedelta.
:param tdelta: the timedelta instance.
:return: the offset for ``tdelta``.
"""
offset_s = tdelta.total_seconds()
offset_h = int(offset_s / 3600)
offset_m = int((offset_s / 60) % 60)
offset_t = time(abs(offset_h), abs(offset_m))
operator = '+' if offset_s > 0 else '-'
offset = offset_t.strftime('{}%H:%M'.format(operator))
return offset
def _datetime_utc_inst(obj: str, pattern: str) -> datetime:
"""
Return a datetime instance with UTC timezone info.
:param obj: a datetime in RFC3339 format.
:param pattern: the datetime pattern that is used.
:return: a datetime instance with timezone info.
"""
dattim_str = obj[0:-1]
dattim_obj = datetime.strptime(dattim_str, pattern)
return _new_datetime(dattim_obj.date(), dattim_obj.time(), timezone.utc)
def _datetime_offset_inst(obj: str, pattern: str) -> datetime:
"""
Return a datetime instance with timezone info.
:param obj: a datetime in RFC3339 format.
:param pattern: the datetime pattern that is used.
:return: a datetime instance with timezone info.
"""
dat_str, tim_str = obj.split('T')
splitter, factor = ('+', 1) if '+' in tim_str else ('-', -1)
naive_tim_str, offset = tim_str.split(splitter)
naive_dattim_str = '{}T{}'.format(dat_str, naive_tim_str)
dattim_obj = datetime.strptime(naive_dattim_str, pattern)
hrs_str, mins_str = offset.split(':')
hrs = int(hrs_str) * factor
mins = int(mins_str) * factor
tz = timezone(offset=timedelta(hours=hrs, minutes=mins))
return _new_datetime(dattim_obj.date(), dattim_obj.time(), tz)
def _new_datetime(date_inst: date, time_inst: time, tzinfo: timezone) \
-> datetime:
"""
Return a datetime instance from a date, time and timezone.
This function was required due to the missing argument for tzinfo under the
Linux Python distribution.
:param date_inst: the date.
:param time_inst: the time.
:param tzinfo: the Timezone.
:return: a combined datetime instance.
"""
return datetime.combine(date_inst, time_inst).replace(tzinfo=tzinfo)

View File

@@ -0,0 +1,113 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functionality for dumping stuff to json.
"""
import json
from typing import Optional, Dict
from jsons._cache import clear
from jsons._common_impl import StateHolder
from jsons._extra_impl import announce_class
from jsons._lizers_impl import get_serializer
from jsons.exceptions import SerializationError
def dump(obj: object,
cls: Optional[type] = None,
*,
strict: bool = False,
fork_inst: Optional[type] = StateHolder,
**kwargs) -> object:
"""
Serialize the given ``obj`` to a JSON equivalent type (e.g. dict, list,
int, ...).
The way objects are serialized can be finetuned by setting serializer
functions for the specific type using ``set_serializer``.
You can also provide ``cls`` to specify that ``obj`` needs to be serialized
as if it was of type ``cls`` (meaning to only take into account attributes
from ``cls``). The type ``cls`` must have a ``__slots__`` defined. Any type
will do, but in most cases you may want ``cls`` to be a base class of
``obj``.
:param obj: a Python instance of any sort.
:param cls: if given, ``obj`` will be dumped as if it is of type ``type``.
:param strict: a bool to determine if the serializer should be strict
(i.e. only dumping stuff that is known to ``cls``).
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:param kwargs: the keyword args are passed on to the serializer function.
:return: the serialized obj as a JSON type.
"""
cls_ = cls or obj.__class__
serializer = get_serializer(cls_, fork_inst)
# Is this the initial call or a nested?
initial = kwargs.get('_initial', True)
kwargs_ = {
'fork_inst': fork_inst,
'_initial': False,
'strict': strict,
**kwargs
}
announce_class(cls_, fork_inst=fork_inst)
return _do_dump(obj, serializer, cls, initial, kwargs_)
def _do_dump(obj, serializer, cls, initial, kwargs):
try:
result = serializer(obj, cls=cls, **kwargs)
if initial:
clear()
return result
except Exception as err:
clear()
raise SerializationError(str(err)) from err
def dumps(obj: object,
jdkwargs: Optional[Dict[str, object]] = None,
*args,
**kwargs) -> str:
"""
Extend ``json.dumps``, allowing any Python instance to be dumped to a
string. Any extra (keyword) arguments are passed on to ``json.dumps``.
:param obj: the object that is to be dumped to a string.
:param jdkwargs: extra keyword arguments for ``json.dumps`` (not
``jsons.dumps``!)
:param args: extra arguments for ``jsons.dumps``.
:param kwargs: Keyword arguments that are passed on through the
serialization process.
passed on to the serializer function.
:return: ``obj`` as a ``str``.
"""
jdkwargs = jdkwargs or {}
dumped = dump(obj, *args, **kwargs)
return json.dumps(dumped, **jdkwargs)
def dumpb(obj: object,
encoding: str = 'utf-8',
jdkwargs: Optional[Dict[str, object]] = None,
*args,
**kwargs) -> bytes:
"""
Extend ``json.dumps``, allowing any Python instance to be dumped to bytes.
Any extra (keyword) arguments are passed on to ``json.dumps``.
:param obj: the object that is to be dumped to bytes.
:param encoding: the encoding that is used to transform to bytes.
:param jdkwargs: extra keyword arguments for ``json.dumps`` (not
``jsons.dumps``!)
:param args: extra arguments for ``jsons.dumps``.
:param kwargs: Keyword arguments that are passed on through the
serialization process.
passed on to the serializer function.
:return: ``obj`` as ``bytes``.
"""
jdkwargs = jdkwargs or {}
dumped_dict = dump(obj, *args, **kwargs)
dumped_str = json.dumps(dumped_dict, **jdkwargs)
return dumped_str.encode(encoding=encoding)

View File

@@ -0,0 +1,53 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains implementations that do not directly touch the core of
jsons.
"""
from typing import Optional
from jsons._cache import cached
from jsons._common_impl import StateHolder, get_class_name
def suppress_warnings(
do_suppress: Optional[bool] = True,
fork_inst: Optional[type] = StateHolder):
"""
Suppress (or stop suppressing) warnings altogether.
:param do_suppress: if ``True``, warnings will be suppressed from now on.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: None.
"""
fork_inst._suppress_warnings = do_suppress
def suppress_warning(
code: str,
fork_inst: Optional[type] = StateHolder):
"""
Suppress a specific warning that corresponds to the given code (see the
warning).
:param code: the code of the warning that is to be suppressed.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: None.
"""
fork_inst._suppressed_warnings |= {code}
@cached
def announce_class(
cls: type,
cls_name: Optional[str] = None,
fork_inst: type = StateHolder):
"""
Announce the given cls to jsons to allow jsons to deserialize a verbose
dump into that class.
:param cls: the class that is to be announced.
:param cls_name: a custom name for that class.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: None.
"""
cls_name = cls_name or get_class_name(cls, fully_qualified=True)
fork_inst._announced_classes[cls] = cls_name
fork_inst._announced_classes[cls_name] = cls

View File

@@ -0,0 +1,38 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains the implementation of ``fork()``.
"""
from typing import Type, Optional
from jsons._common_impl import StateHolder, get_class_name, T
def fork(
fork_inst: Type[T] = StateHolder,
name: Optional[str] = None) -> Type[T]:
"""
Fork from the given ``StateHolder`` to create a separate "branch" of
serializers and deserializers.
:param fork_inst: The ``StateHolder`` on which the new fork is based.
:param name: The ``__name__`` of the new ``type``.
:return: A "fork inst" that can be used to separately store
(de)serializers from the regular ``StateHolder``.
"""
fork_inst._fork_counter += 1
if name:
class_name = name
else:
class_name = '{}_fork{}'.format(
get_class_name(fork_inst),
fork_inst._fork_counter
)
result = type(class_name, (fork_inst,), {})
result._classes_serializers = fork_inst._classes_serializers.copy()
result._classes_deserializers = fork_inst._classes_deserializers.copy()
result._serializers = fork_inst._serializers.copy()
result._deserializers = fork_inst._deserializers.copy()
result._fork_counter = 0
result._suppress_warnings = fork_inst._suppress_warnings
result._suppressed_warnings = fork_inst._suppressed_warnings.copy()
return result

View File

@@ -0,0 +1,50 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functions that can be used to transform keys of
dictionaries.
"""
import re
def camelcase(str_: str) -> str:
"""
Return ``s`` in camelCase.
:param str_: the string that is to be transformed.
:return: a string in camelCase.
"""
str_ = str_.replace('-', '_')
splitted = str_.split('_')
if len(splitted) > 1:
str_ = ''.join([x.title() for x in splitted])
return str_[0].lower() + str_[1:]
def snakecase(str_: str) -> str:
"""
Return ``s`` in snake_case.
:param str_: the string that is to be transformed.
:return: a string in snake_case.
"""
str_ = str_.replace('-', '_')
str_ = str_[0].lower() + str_[1:]
return re.sub(r'([a-z])([A-Z])', '\\1_\\2', str_).lower()
def pascalcase(str_: str) -> str:
"""
Return ``s`` in PascalCase.
:param str_: the string that is to be transformed.
:return: a string in PascalCase.
"""
camelcase_str = camelcase(str_)
return camelcase_str[0].upper() + camelcase_str[1:]
def lispcase(str_: str) -> str:
"""
Return ``s`` in lisp-case.
:param str_: the string that is to be transformed.
:return: a string in lisp-case.
"""
return snakecase(str_).replace('_', '-')

View File

@@ -0,0 +1,159 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functionality for setting and getting serializers and
deserializers.
"""
from typing import Optional, Dict, Sequence, Union
from jsons._cache import cached
from jsons._common_impl import StateHolder, get_class_name
from jsons._compatibility_impl import get_naked_class
def set_serializer(
func: callable,
cls: Union[type, Sequence[type]],
high_prio: bool = True,
fork_inst: type = StateHolder) -> None:
"""
Set a serializer function for the given type. You may override the default
behavior of ``jsons.load`` by setting a custom serializer.
The ``func`` argument must take one argument (i.e. the object that is to be
serialized) and also a ``kwargs`` parameter. For example:
>>> def func(obj, **kwargs):
... return dict()
You may ask additional arguments between ``cls`` and ``kwargs``.
:param func: the serializer function.
:param cls: the type or sequence of types this serializer can handle.
:param high_prio: determines the order in which is looked for the callable.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: None.
"""
if isinstance(cls, Sequence):
for cls_ in cls:
set_serializer(func, cls_, high_prio, fork_inst)
elif cls:
index = 0 if high_prio else len(fork_inst._classes_serializers)
fork_inst._classes_serializers.insert(index, cls)
cls_name = get_class_name(cls, fully_qualified=True)
fork_inst._serializers[cls_name.lower()] = func
else:
fork_inst._serializers['nonetype'] = func
def set_deserializer(
func: callable,
cls: Union[type, Sequence[type]],
high_prio: bool = True,
fork_inst: type = StateHolder) -> None:
"""
Set a deserializer function for the given type. You may override the
default behavior of ``jsons.dump`` by setting a custom deserializer.
The ``func`` argument must take two arguments (i.e. the dict containing the
serialized values and the type that the values should be deserialized into)
and also a ``kwargs`` parameter. For example:
>>> def func(dict_, cls, **kwargs):
... return cls()
You may ask additional arguments between ``cls`` and ``kwargs``.
:param func: the deserializer function.
:param cls: the type or sequence of types this serializer can handle.
:param high_prio: determines the order in which is looked for the callable.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: None.
"""
if isinstance(cls, Sequence):
for cls_ in cls:
set_deserializer(func, cls_, high_prio, fork_inst)
elif cls:
index = 0 if high_prio else len(fork_inst._classes_deserializers)
fork_inst._classes_deserializers.insert(index, cls)
cls_name = get_class_name(cls, fully_qualified=True)
fork_inst._deserializers[cls_name.lower()] = func
else:
fork_inst._deserializers['nonetype'] = func
@cached
def get_serializer(
cls: type,
fork_inst: Optional[type] = StateHolder) -> callable:
"""
Return the serializer function that would be used for the given ``cls``.
:param cls: the type for which a serializer is to be returned.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: a serializer function.
"""
serializer = _get_lizer(cls, fork_inst._serializers,
fork_inst._classes_serializers, fork_inst)
return serializer
@cached
def get_deserializer(
cls: type,
fork_inst: Optional[type] = StateHolder) -> callable:
"""
Return the deserializer function that would be used for the given ``cls``.
:param cls: the type for which a deserializer is to be returned.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: a deserializer function.
"""
deserializer = _get_lizer(cls, fork_inst._deserializers,
fork_inst._classes_deserializers, fork_inst)
return deserializer
def _get_lizer(
cls: type,
lizers: Dict[str, callable],
classes_lizers: list,
fork_inst: type,
recursive: bool = False) -> callable:
cls_name = get_class_name(cls, str.lower, fully_qualified=True)
lizer = (lizers.get(cls_name, None)
or _get_lizer_by_parents(cls, lizers, classes_lizers, fork_inst))
if not lizer and not recursive and hasattr(cls, '__supertype__'):
return _get_lizer(cls.__supertype__, lizers,
classes_lizers, fork_inst, True)
return lizer
def _get_lizer_by_parents(
cls: type,
lizers: Dict[str, callable],
classes_lizers: list,
fork_inst: type) -> callable:
result = None
parents = _get_parents(cls, classes_lizers)
if parents:
pname = get_class_name(parents[0], str.lower, fully_qualified=True)
result = lizers[pname]
return result
def _get_parents(cls: type, lizers: list) -> list:
"""
Return a list of serializers or deserializers that can handle a parent
of ``cls``.
:param cls: the type that
:param lizers: a list of serializers or deserializers.
:return: a list of serializers or deserializers.
"""
parents = []
naked_cls = get_naked_class(cls)
for cls_ in lizers:
try:
if issubclass(naked_cls, cls_):
parents.append(cls_)
except (TypeError, AttributeError):
pass # Some types do not support `issubclass` (e.g. Union).
return parents

View File

@@ -0,0 +1,219 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functionality for loading stuff from json.
"""
import json
from json import JSONDecodeError
from typing import Optional, Dict, Callable, Tuple, Any, Type
from jsons._cache import clear
from jsons._common_impl import (
StateHolder,
get_cls_from_str,
get_class_name,
get_cls_and_meta,
determine_precedence,
VALID_TYPES,
T,
can_match_with_none
)
from jsons._lizers_impl import get_deserializer
from jsons._validation import validate
from jsons.exceptions import DeserializationError, JsonsError, DecodeError
def load(
json_obj: object,
cls: Optional[Type[T]] = None,
*,
strict: bool = False,
fork_inst: Optional[type] = StateHolder,
attr_getters: Optional[Dict[str, Callable[[], object]]] = None,
**kwargs) -> T:
"""
Deserialize the given ``json_obj`` to an object of type ``cls``. If the
contents of ``json_obj`` do not match the interface of ``cls``, a
DeserializationError is raised.
If ``json_obj`` contains a value that belongs to a custom class, there must
be a type hint present for that value in ``cls`` to let this function know
what type it should deserialize that value to.
**Example**:
>>> from typing import List
>>> import jsons
>>> class Person:
... # No type hint required for name
... def __init__(self, name):
... self.name = name
>>> class Family:
... # Person is a custom class, use a type hint
... def __init__(self, persons: List[Person]):
... self.persons = persons
>>> loaded = jsons.load({'persons': [{'name': 'John'}]}, Family)
>>> loaded.persons[0].name
'John'
If no ``cls`` is given, a dict is simply returned, but contained values
(e.g. serialized ``datetime`` values) are still deserialized.
If `strict` mode is off and the type of `json_obj` exactly matches `cls`
then `json_obj` is simply returned.
:param json_obj: the dict that is to be deserialized.
:param cls: a matching class of which an instance should be returned.
:param strict: a bool to determine if the deserializer should be strict
(i.e. fail on a partially deserialized `json_obj` or on `None`).
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:param attr_getters: a ``dict`` that may hold callables that return values
for certain attributes.
:param kwargs: the keyword args are passed on to the deserializer function.
:return: an instance of ``cls`` if given, a dict otherwise.
"""
_check_for_none(json_obj, cls)
if _should_skip(json_obj, cls, strict):
validate(json_obj, cls, fork_inst)
return json_obj
if isinstance(cls, str):
cls = get_cls_from_str(cls, json_obj, fork_inst)
original_cls = cls
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)
# Is this the initial call or a nested?
initial = kwargs.get('_initial', True)
kwargs_ = {
'meta_hints': meta_hints, # Overridable by kwargs.
**kwargs,
'strict': strict,
'fork_inst': fork_inst,
'attr_getters': attr_getters,
'_initial': False,
'_inferred_cls': cls is not original_cls,
}
return _do_load(json_obj, deserializer, cls, initial, **kwargs_)
def _do_load(json_obj: object,
deserializer: callable,
cls: type,
initial: bool,
**kwargs):
cls_name = get_class_name(cls, fully_qualified=True)
if deserializer is None:
raise DeserializationError('No deserializer for type "{}"'.format(cls_name), json_obj, cls)
try:
result = deserializer(json_obj, cls, **kwargs)
validate(result, cls, kwargs['fork_inst'])
except Exception as err:
clear()
if isinstance(err, JsonsError):
raise
message = 'Could not deserialize value "{}" into "{}". {}'.format(json_obj, cls_name, err)
raise DeserializationError(message, json_obj, cls) from err
else:
if initial:
# Clear all lru caches right before returning the initial call.
clear()
return result
def loads(
str_: str,
cls: Optional[Type[T]] = None,
jdkwargs: Optional[Dict[str, object]] = None,
*args,
**kwargs) -> T:
"""
Extend ``json.loads``, allowing a string to be loaded into a dict or a
Python instance of type ``cls``. Any extra (keyword) arguments are passed
on to ``json.loads``.
:param str_: the string that is to be loaded.
:param cls: a matching class of which an instance should be returned.
:param jdkwargs: extra keyword arguments for ``json.loads`` (not
``jsons.loads``!)
:param args: extra arguments for ``jsons.loads``.
:param kwargs: extra keyword arguments for ``jsons.loads``.
:return: a JSON-type object (dict, str, list, etc.) or an instance of type
``cls`` if given.
"""
jdkwargs = jdkwargs or {}
try:
obj = json.loads(str_, **jdkwargs)
except JSONDecodeError as err:
raise DecodeError('Could not load a dict; the given string is not '
'valid JSON.', str_, cls, err) from err
else:
return load(obj, cls, *args, **kwargs)
def loadb(
bytes_: bytes,
cls: Optional[Type[T]] = None,
encoding: str = 'utf-8',
jdkwargs: Optional[Dict[str, object]] = None,
*args,
**kwargs) -> T:
"""
Extend ``json.loads``, allowing bytes to be loaded into a dict or a Python
instance of type ``cls``. Any extra (keyword) arguments are passed on to
``json.loads``.
:param bytes_: the bytes that are to be loaded.
:param cls: a matching class of which an instance should be returned.
:param encoding: the encoding that is used to transform from bytes.
:param jdkwargs: extra keyword arguments for ``json.loads`` (not
``jsons.loads``!)
:param args: extra arguments for ``jsons.loads``.
:param kwargs: extra keyword arguments for ``jsons.loads``.
:return: a JSON-type object (dict, str, list, etc.) or an instance of type
``cls`` if given.
"""
if not isinstance(bytes_, bytes):
raise DeserializationError('loadb accepts bytes only, "{}" was given'
.format(type(bytes_)), bytes_, cls)
jdkwargs = jdkwargs or {}
str_ = bytes_.decode(encoding=encoding)
return loads(str_, cls, jdkwargs=jdkwargs, *args, **kwargs)
def _check_and_get_cls_and_meta_hints(
json_obj: object,
cls: type,
fork_inst: type,
inferred_cls: bool) -> Tuple[type, Optional[dict]]:
# Check if json_obj is of a valid type and return the cls.
if type(json_obj) not in VALID_TYPES:
invalid_type = get_class_name(type(json_obj), fully_qualified=True)
valid_types = [get_class_name(typ, fully_qualified=True)
for typ in VALID_TYPES]
msg = ('Invalid type: "{}", only arguments of the following types are '
'allowed: {}'.format(invalid_type, ", ".join(valid_types)))
raise DeserializationError(msg, json_obj, cls)
cls_from_meta, meta = get_cls_and_meta(json_obj, fork_inst)
meta_hints = meta.get('classes', {}) if meta else {}
return determine_precedence(
cls, cls_from_meta, type(json_obj), inferred_cls), meta_hints
def _should_skip(json_obj: object, cls: type, strict: bool):
return (not strict and type(json_obj) == cls) or cls is Any
def _check_for_none(json_obj: object, cls: type):
# Check if the json_obj is None and whether or not that is fine.
if json_obj is None and not can_match_with_none(cls):
cls_name = get_class_name(cls)
raise DeserializationError(
message='NoneType cannot be deserialized into {}'.format(cls_name),
source=json_obj,
target=cls)

View File

@@ -0,0 +1,82 @@
"""
PRIVATE MODULE: do not import (from) it directly.
Functionality for processing iterables in parallel.
"""
from multiprocessing import Process, Manager
from typing import List, Callable, Union
from typish import Something
Subscriptable = Something['__getitem__': Callable[[int], object]]
def multi_task(
func: Callable,
obj: Subscriptable,
tasks: int,
task_type: type,
*args,
**kwargs):
result = _get_list_to_fill(obj, task_type)
tasks_instances = _start_tasks(tasks=tasks, task_type=task_type, func=func,
list_to_fill=result, obj=obj, args=args,
kwargs=kwargs)
for task in tasks_instances:
task.join()
return list(result)
def _get_list_to_fill(obj: list, task_type: type) -> Union[list, Manager]:
# Return a list or manager that contains enough spots to fill.
result = [0] * len(obj)
if issubclass(task_type, Process):
manager = Manager()
result = manager.list(result)
return result
def _start_tasks(
tasks: int,
task_type: type,
func: Callable,
list_to_fill: list,
obj: Subscriptable,
args,
kwargs) -> List[Something['join': Callable[[], None]]]:
# Start the tasks and return their instances so they can be joined.
tasks_instances = []
tasks_used = min(tasks, len(obj))
tasks_left = tasks - tasks_used or 1
# Divide the list in parts.
slice_size = int(len(obj) / tasks_used)
rest_size = len(obj) % tasks_used
for i in range(tasks_used):
start = i * slice_size
end = (i + 1) * slice_size
if i == tasks_used - 1:
end += rest_size
task = task_type(
target=_fill,
args=(func, list_to_fill, obj, start, end, tasks_left, args, kwargs))
task.start()
tasks_instances.append(task)
return tasks_instances
def _fill(
func,
list_to_fill: list,
obj: Subscriptable,
start: int,
end: int,
tasks: int,
args,
kwargs):
# Fill the given list with results from func.
for i_ in range(start, end):
loaded = func(obj[i_], tasks=tasks, *args, **kwargs)
list_to_fill[i_] = loaded

View File

@@ -0,0 +1,15 @@
__title__ = 'jsons'
__version__ = '1.6.3'
__author__ = 'Ramon Hagenaars'
__author_email__ = 'ramon.hagenaars@gmail.com'
__description__ = 'For serializing Python objects to JSON (dicts) and back'
__url__ = 'https://github.com/ramonhagenaars/jsons'
__license__ = 'MIT'
__python_versions__ = [
'3.5',
'3.6',
'3.7',
'3.8',
'3.9',
'3.10',
]

View File

@@ -0,0 +1,40 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functionality for loading stuff from json.
"""
from typing import Type, List, Any, Dict, Callable
from jsons._common_impl import T
from jsons._dump_impl import dump
from jsons._load_impl import load
def transform(
obj: object,
cls: Type[T],
*,
mapper: Callable[[Dict[str, Any]], Dict[str, Any]] = None,
dump_cls: type = None,
dump_args: List[Any] = None,
dump_kwargs: List[Dict[str, Any]] = None,
**kwargs) -> T:
"""
Transform the given ``obj`` to an instance of ``cls``.
:param obj: the object that is to be transformed into a type of ``cls``.
:param cls: the type that ``obj`` is to be transformed into.
:param mapper: a callable that takes the dumped dict and returns a mapped
dict right before it is loaded into ``cls``.
:param dump_cls: the ``cls`` parameter that is given to ``dump``.
:param dump_args: the ``args`` parameter that is given to ``dump``.
:param dump_kwargs: the ``kwargs`` parameter that is given to ``dump``.
:param kwargs: any keyword arguments that are given to ``load``.
:return: an instance of ``cls``.
"""
dump_args_ = dump_args or []
dump_kwargs_ = dump_kwargs or {}
dumped = dump(obj, dump_cls, *dump_args_, **dump_kwargs_)
mapper_ = mapper or (lambda x: x)
dumped_mapped = mapper_(dumped)
return load(dumped_mapped, cls, **kwargs)

View File

@@ -0,0 +1,75 @@
"""
PRIVATE MODULE: do not import (from) it directly.
This module contains functionality for validating objects.
"""
from typing import Union, Sequence, Callable
from jsons._cache import cached
from jsons._common_impl import StateHolder, get_class_name
from jsons._lizers_impl import _get_lizer
from jsons.exceptions import ValidationError
def set_validator(
func: Callable[[object], bool],
cls: Union[type, Sequence[type]],
*,
fork_inst: type = StateHolder) -> None:
"""
Set a validator function for the given ``cls``. The function should accept
an instance of the type it should validate and must return ``False`` or
raise any exception in case of a validation failure.
:param func: the function that takes an instance of type ``cls`` and
returns a bool (``True`` if the validation was successful).
:param cls: the type or types that ``func`` is able to validate.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: None.
"""
if isinstance(cls, Sequence):
for cls_ in cls:
set_validator(func, cls=cls_, fork_inst=fork_inst)
else:
cls_name = get_class_name(cls, fully_qualified=True)
fork_inst._validators[cls_name.lower()] = func
fork_inst._classes_validators.append(cls)
@cached
def get_validator(
cls: type,
fork_inst: type = StateHolder) -> callable:
"""
Return the validator function that would be used for the given ``cls``.
:param cls: the type for which a deserializer is to be returned.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: a validator function.
"""
return _get_lizer(cls, fork_inst._validators,
fork_inst._classes_validators, fork_inst)
def validate(
obj: object,
cls: type,
fork_inst: type = StateHolder) -> None:
"""
Validate the given ``obj`` with the validator that was registered for
``cls``. Raises a ``ValidationError`` if the validation failed.
:param obj: the object that is to be validated.
:param cls: the type of which the validator function was registered.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:return: None.
"""
validator = get_validator(cls, fork_inst)
result = True
msg = 'Validation failed.'
if validator:
try:
result = validator(obj)
except Exception as err:
if err.args:
msg = err.args[0]
result = False
if not result:
raise ValidationError(msg)

View File

@@ -0,0 +1,7 @@
from jsons.classes.json_serializable import JsonSerializable
from jsons.classes.verbosity import Verbosity
__all__ = [
JsonSerializable.__name__,
Verbosity.__name__
]

View File

@@ -0,0 +1,217 @@
from typing import Optional, Type
from jsons._common_impl import StateHolder, T
from jsons._dump_impl import dump, dumps, dumpb
from jsons._fork_impl import fork
from jsons._lizers_impl import set_serializer, set_deserializer
from jsons._load_impl import load, loads, loadb
class JsonSerializable(StateHolder):
"""
This class offers an alternative to using the ``jsons.load`` and
``jsons.dump`` methods. An instance of a class that inherits from
``JsonSerializable`` has the ``json`` property, which value is equivalent
to calling ``jsons.dump`` on that instance. Furthermore, you can call
``from_json`` on that class, which is equivalent to calling ``json.load``
with that class as an argument.
"""
@classmethod
def fork(cls, name: Optional[str] = None) -> Type['JsonSerializable']:
"""
Create a 'fork' of ``JsonSerializable``: a new ``type`` with a separate
configuration of serializers and deserializers.
:param name: the ``__name__`` of the new ``type``.
:return: a new ``type`` based on ``JsonSerializable``.
"""
return fork(cls, name=name)
@classmethod
def with_dump(cls, fork: Optional[bool] = False, **kwargs) \
-> Type['JsonSerializable']:
"""
Return a class (``type``) that is based on JsonSerializable with the
``dump`` method being automatically provided the given ``kwargs``.
**Example:**
>>> custom_serializable = JsonSerializable\
.with_dump(key_transformer=KEY_TRANSFORMER_CAMELCASE)
>>> class Person(custom_serializable):
... def __init__(self, my_name):
... self.my_name = my_name
>>> p = Person('John')
>>> p.json
{'myName': 'John'}
:param kwargs: the keyword args that are automatically provided to the
``dump`` method.
:param fork: determines that a new fork is to be created.
:return: a class with customized behavior.
"""
def _wrapper(inst, **kwargs_):
return dump(inst, **{**kwargs_, **kwargs})
type_ = cls.fork() if fork else cls
type_.dump = _wrapper
return type_
@classmethod
def with_load(cls, fork: Optional[bool] = False, **kwargs) \
-> Type['JsonSerializable']:
"""
Return a class (``type``) that is based on JsonSerializable with the
``load`` method being automatically provided the given ``kwargs``.
**Example:**
>>> custom_serializable = JsonSerializable\
.with_load(key_transformer=KEY_TRANSFORMER_SNAKECASE)
>>> class Person(custom_serializable):
... def __init__(self, my_name):
... self.my_name = my_name
>>> p_json = {'myName': 'John'}
>>> p = Person.from_json(p_json)
>>> p.my_name
'John'
:param kwargs: the keyword args that are automatically provided to the
``load`` method.
:param fork: determines that a new fork is to be created.
:return: a class with customized behavior.
"""
@classmethod
def _wrapper(cls_, inst, **kwargs_):
return load(inst, cls_, fork_inst=cls_, **{**kwargs_, **kwargs})
type_ = cls.fork() if fork else cls
type_.load = _wrapper
return type_
@property
def json(self) -> object:
"""
See ``jsons.dump``.
:return: this instance in a JSON representation (dict).
"""
return self.dump()
def __str__(self) -> str:
"""
See ``jsons.dumps``.
:return: this instance as a JSON string.
"""
return self.dumps()
@classmethod
def from_json(cls: Type[T], json_obj: object, **kwargs) -> T:
"""
See ``jsons.load``.
:param json_obj: a JSON representation of an instance of the inheriting
class
:param kwargs: the keyword args are passed on to the deserializer
function.
:return: an instance of the inheriting class.
"""
return cls.load(json_obj, **kwargs)
def dump(self, **kwargs) -> object:
"""
See ``jsons.dump``.
:param kwargs: the keyword args are passed on to the serializer
function.
:return: this instance in a JSON representation (dict).
"""
return dump(self, fork_inst=self.__class__, **kwargs)
@classmethod
def load(cls: Type[T], json_obj: object, **kwargs) -> T:
"""
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)
def dumps(self, **kwargs) -> str:
"""
See ``jsons.dumps``.
:param kwargs: the keyword args are passed on to the serializer
function.
:return: this instance as a JSON string.
"""
return dumps(self, fork_inst=self.__class__, **kwargs)
@classmethod
def loads(cls: Type[T], json_obj: str, **kwargs) -> T:
"""
See ``jsons.loads``.
: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 loads(json_obj, cls, fork_inst=cls, **kwargs)
def dumpb(self, **kwargs) -> bytes:
"""
See ``jsons.dumpb``.
:param kwargs: the keyword args are passed on to the serializer
function.
:return: this instance as a JSON string.
"""
return dumpb(self, fork_inst=self.__class__, **kwargs)
@classmethod
def loadb(cls: Type[T], json_obj: bytes, **kwargs) -> T:
"""
See ``jsons.loadb``.
: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 loadb(json_obj, cls, fork_inst=cls, **kwargs)
@classmethod
def set_serializer(cls: Type[T],
func: callable,
cls_: type,
high_prio: Optional[bool] = True,
fork: Optional[bool] = False) -> T:
"""
See ``jsons.set_serializer``.
:param func: the serializer function.
:param cls_: the type this serializer can handle.
:param high_prio: determines the order in which is looked for the
callable.
:param fork: determines that a new fork is to be created.
:return: the type on which this method is invoked or its fork.
"""
type_ = cls.fork() if fork else cls
set_serializer(func, cls_, high_prio, type_)
return type_
@classmethod
def set_deserializer(cls: Type[T],
func: callable,
cls_: type,
high_prio: Optional[bool] = True,
fork: Optional[bool] = False) -> T:
"""
See ``jsons.set_deserializer``.
:param func: the deserializer function.
:param cls_: the type this serializer can handle.
:param high_prio: determines the order in which is looked for the
callable.
:param fork: determines that a new fork is to be created.
:return: the type on which this method is invoked or its fork.
"""
type_ = cls.fork() if fork else cls
set_deserializer(func, cls_, high_prio, type_)
return type_

View File

@@ -0,0 +1,32 @@
try:
from enum import Flag
except ImportError: # pragma: no cover
from jsons._compatibility_impl import Flag
class Verbosity(Flag):
"""
An enum that defines the level of verbosity of the serialization of an
object.
"""
WITH_NOTHING = 0
WITH_CLASS_INFO = 10
WITH_DUMP_TIME = 20
WITH_EVERYTHING = WITH_CLASS_INFO | WITH_DUMP_TIME
@staticmethod
def from_value(value: any) -> 'Verbosity':
"""
Return a ``Verbosity`` instance from the given value.
:param value:
:return: a ``Verbosity`` instance corresponding to ``value``.
"""
if isinstance(value, Verbosity):
return value
if value in (False, None):
return Verbosity.WITH_NOTHING
if value is True:
return Verbosity.WITH_EVERYTHING
if value:
return Verbosity.WITH_EVERYTHING
return Verbosity.WITH_NOTHING

View File

@@ -0,0 +1,188 @@
"""
This module contains decorators that facilitate the `jsons` functions in an
alternative fashion.
"""
from inspect import signature, Parameter, isawaitable, iscoroutinefunction
from jsons import JsonSerializable, dump, load, loads, loadb, dumps, dumpb
from jsons.exceptions import InvalidDecorationError
def loaded(
parameters=True,
returnvalue=True,
fork_inst=JsonSerializable,
loader=load,
**kwargs):
"""
Return a decorator that can call `jsons.load` upon all parameters and the
return value of the decorated function.
**Example**:
>>> from datetime import datetime
>>> @loaded()
... def func(arg: datetime) -> datetime:
... # arg is now of type datetime.
... return '2018-10-04T21:57:00Z' # This will become a datetime.
>>> res = func('2018-10-04T21:57:00Z')
>>> type(res).__name__
'datetime'
:param parameters: determines whether parameters should be taken into
account.
:param returnvalue: determines whether the return value should be taken
into account.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:param kwargs: any keyword arguments that should be passed on to
`jsons.load`
:param loader: the load function which must be one of (``load``,
``loads``, ``loadb``)
:return: a decorator that can be placed on a function.
"""
if loader not in (load, loads, loadb):
raise InvalidDecorationError("The 'loader' argument must be one of: "
"jsons.load, jsons.loads, jsons.loadb")
return _get_decorator(parameters, returnvalue, fork_inst, loader, kwargs)
def dumped(
parameters=True,
returnvalue=True,
fork_inst=JsonSerializable,
dumper=dump,
**kwargs):
"""
Return a decorator that can call `jsons.dump` upon all parameters and the
return value of the decorated function.
**Example**:
>>> from datetime import datetime
>>> @dumped()
... def func(arg):
... # arg is now of type str.
... return datetime.now()
>>> res = func(datetime.now())
>>> type(res).__name__
'str'
:param parameters: determines whether parameters should be taken into
account.
:param returnvalue: determines whether the return value should be taken
into account.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
:param kwargs: any keyword arguments that should be passed on to
`jsons.dump`
:param dumper: the dump function which must be one of (``dump``,
``dumps``, ``dumpb``)
:return: a decorator that can be placed on a function.
"""
if dumper not in (dump, dumps, dumpb):
raise InvalidDecorationError("The 'dumper' argument must be one of: "
"jsons.dump, jsons.dumps, jsons.dumpb")
return _get_decorator(parameters, returnvalue, fork_inst, dumper, kwargs)
def _get_decorator(parameters, returnvalue, fork_inst, mapper, mapper_kwargs):
def _decorator(decorated):
_validate_decoration(decorated, fork_inst)
args = [decorated, parameters, returnvalue,
fork_inst, mapper, mapper_kwargs]
wrapper = (_get_async_wrapper(*args) if iscoroutinefunction(decorated)
else _get_wrapper(*args))
return wrapper
return _decorator
def _get_wrapper(
decorated,
parameters,
returnvalue,
fork_inst,
mapper,
mapper_kwargs):
def _wrapper(*args, **kwargs):
result = _run_decorated(decorated, mapper if parameters else None,
fork_inst, args, kwargs, mapper_kwargs)
if returnvalue:
result = _map_returnvalue(result, decorated, fork_inst, mapper,
mapper_kwargs)
return result
return _wrapper
def _get_async_wrapper(
decorated,
parameters,
returnvalue,
fork_inst,
mapper,
mapper_kwargs):
async def _async_wrapper(*args, **kwargs):
result = _run_decorated(decorated, mapper if parameters else None,
fork_inst, args, kwargs, mapper_kwargs)
if isawaitable(result):
result = await result
if returnvalue:
result = _map_returnvalue(result, decorated, fork_inst, mapper,
mapper_kwargs)
return result
return _async_wrapper
def _get_params_sig(args, func):
sig = signature(func)
params = sig.parameters
param_names = [param_name for param_name in params]
result = [(args[i], params[param_names[i]]) for i in range(len(args))]
return result
def _map_args(args, decorated, fork_inst, mapper, mapper_kwargs):
params_sig = _get_params_sig(args, decorated)
new_args = []
for arg, sig in params_sig:
if sig.name in ('self', 'cls') and hasattr(arg, decorated.__name__):
# `decorated` is a method and arg is either `self` or `cls`.
new_arg = arg
else:
cls = sig.annotation if sig.annotation != Parameter.empty else None
new_arg = mapper(arg, cls=cls, fork_inst=fork_inst,
**mapper_kwargs)
new_args.append(new_arg)
return new_args
def _map_returnvalue(returnvalue, decorated, fork_inst, mapper, mapper_kwargs):
return_annotation = signature(decorated).return_annotation
cls = return_annotation if return_annotation != Parameter.empty else None
result = mapper(returnvalue, cls=cls, fork_inst=fork_inst, **mapper_kwargs)
return result
def _run_decorated(decorated, mapper, fork_inst, args, kwargs, mapper_kwargs):
new_args = args
if mapper:
new_args = _map_args(args, decorated, fork_inst, mapper, mapper_kwargs)
result = decorated(*new_args, **kwargs)
return result
def _validate_decoration(decorated, fork_inst):
if isinstance(decorated, staticmethod):
fork_inst._warn('You cannot decorate a static- or classmethod. '
'You can still obtain the desired behavior by '
'decorating your method first and then place '
'@staticmethod/@classmethod on top (switching the '
'order).', 'decorated-static')
raise InvalidDecorationError(
'Cannot decorate a static- or classmethod.')
if isinstance(decorated, type):
raise InvalidDecorationError('Cannot decorate a class.')

View File

@@ -0,0 +1,6 @@
"""
This package contains default deserializers. You can override the
deserialization process of a particular type as follows:
``jsons.set_deserializer(custom_deserializer, SomeClass)``
"""

View File

@@ -0,0 +1,28 @@
from typing import Dict
from jsons._load_impl import load
from jsons.exceptions import DeserializationError
def default_complex_deserializer(obj: Dict[str, float],
cls: type = complex,
**kwargs) -> complex:
"""
Deserialize a dictionary with 'real' and 'imag' keys to a complex number.
:param obj: the dict that is to be deserialized.
:param cls: not used.
:param kwargs: not used.
:return: an instance of ``complex``.
"""
try:
clean_obj = load({'real': obj['real'], 'imag': obj['imag']},
cls=Dict[str, float])
return complex(clean_obj['real'], clean_obj['imag'])
except KeyError as err:
raise AttributeError("Cannot deserialize {} to a complex number, "
"does not contain key '{}'"
.format(obj, err.args[0])) from err
except DeserializationError as err:
raise AttributeError("Cannot deserialize {} to a complex number, "
"cannot cast value {} to float"
.format(obj, err.source)) from err

View File

@@ -0,0 +1,16 @@
from datetime import date
from jsons._datetime_impl import get_datetime_inst, RFC3339_DATE_PATTERN
def default_date_deserializer(obj: str,
cls: type = date,
**kwargs) -> date:
"""
Deserialize a string with an RFC3339 pattern to a date instance.
:param obj: the string that is to be deserialized.
:param cls: not used.
:param kwargs: not used.
:return: a ``datetime.date`` instance.
"""
return get_datetime_inst(obj, RFC3339_DATE_PATTERN).date()

View File

@@ -0,0 +1,24 @@
import datetime
import re
from jsons._datetime_impl import get_datetime_inst, RFC3339_DATETIME_PATTERN
def default_datetime_deserializer(obj: str,
cls: type = datetime,
**kwargs) -> datetime:
"""
Deserialize a string with an RFC3339 pattern to a datetime instance.
:param obj: the string that is to be deserialized.
:param cls: not used.
:param kwargs: not used.
:return: a ``datetime.datetime`` instance.
"""
pattern = RFC3339_DATETIME_PATTERN
if '.' in obj:
pattern += '.%f'
# strptime allows a fraction of length 6, so trip the rest (if exists).
regex_pattern = re.compile(r'(\.[0-9]+)')
frac = regex_pattern.search(obj).group()
obj = obj.replace(frac, frac[0:7])
return get_datetime_inst(obj, pattern)

View File

@@ -0,0 +1,16 @@
from decimal import Decimal
from typing import Optional, Union
def default_decimal_deserializer(obj: Union[str, float, int],
cls: Optional[type] = None,
**kwargs) -> Decimal:
"""
Deserialize a Decimal. Expects a string representation of a number, or
the number itself as a float or int.
:param obj: the string float or int that is to be deserialized.
:param cls: not used.
:param kwargs: any keyword arguments.
:return: the deserialized obj.
"""
return Decimal(obj)

View File

@@ -0,0 +1,32 @@
from collections import defaultdict
from typing import Optional, Callable, Dict
from typish import get_args
from jsons._load_impl import load
def default_defaultdict_deserializer(
obj: dict,
cls: type,
*,
key_transformer: Optional[Callable[[str], str]] = None,
**kwargs) -> dict:
"""
Deserialize a defaultdict.
: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 defaultdict instance.
"""
args = get_args(cls)
default_factory = None
cls_ = Dict
if args:
key, value = get_args(cls)
cls_ = Dict[key, value]
default_factory = value
loaded = load(obj, cls_, key_transformer=key_transformer, **kwargs)
return defaultdict(default_factory, loaded)

View File

@@ -0,0 +1,87 @@
from typing import Callable, Optional, Tuple
from typish import get_args
from jsons._load_impl import load
from jsons.exceptions import DeserializationError
def default_dict_deserializer(
obj: dict,
cls: type,
*,
key_transformer: Optional[Callable[[str], str]] = None,
**kwargs) -> dict:
"""
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.
"""
cls_args = get_args(cls)
obj_, keys_were_hashed = _load_hashed_keys(
obj, cls, cls_args, key_transformer=key_transformer, **kwargs)
return _deserialize(obj_, cls_args, key_transformer, keys_were_hashed, kwargs)
def _load_hashed_keys(
obj: dict,
cls: type,
cls_args: tuple,
**kwargs) -> Tuple[dict, bool]:
# Load any hashed keys and return a copy of the given obj if any hashed
# keys are unpacked.
result = obj
stored_keys = set(obj.get('-keys', set()))
if stored_keys:
# Apparently, there are stored hashed keys, we need to unpack them.
if len(cls_args) != 2:
raise DeserializationError('A detailed type is needed for cls of '
'the form Dict[<type>, <type>] to '
'deserialize a dict with hashed keys.',
obj, cls)
result = {**obj}
key_type = cls_args[0]
for key in stored_keys:
# Get the original (unhashed) key and load it.
original_key = result['-keys'][key]
loaded_key = load(original_key, cls=key_type, **kwargs)
# Replace the hashed key by the loaded key entirely.
result[loaded_key] = result[key]
del result['-keys'][key]
del result[key]
del result['-keys']
return result, len(stored_keys) > 0
def _deserialize(
obj: dict,
cls_args: tuple,
key_transformer: Callable[[str], str],
keys_were_hashed: bool,
kwargs: dict) -> dict:
key_transformer = key_transformer or (lambda key: key)
key_func = key_transformer
kwargs_ = {**kwargs, 'key_transformer': key_transformer}
if len(cls_args) == 2:
cls_k, cls_v = cls_args
kwargs_['cls'] = cls_v
if not keys_were_hashed:
# In case of cls is something like Dict[<key>, <value>], we need to
# ensure that the keys in the result are <key>. If the keys were
# hashed though, they have been loaded already.
kwargs_k = {**kwargs, 'cls': cls_k}
key_func = lambda key: load(key_transformer(key), **kwargs_k)
return {
key_func(key): load(obj[key], **kwargs_)
for key in obj
}

View File

@@ -0,0 +1,32 @@
from enum import EnumMeta
from typing import Optional
def default_enum_deserializer(obj: str,
cls: EnumMeta,
*,
use_enum_name: Optional[bool] = None,
**kwargs) -> object:
"""
Deserialize an enum value to an enum instance. The serialized value can be
either the name or the key of an enum entry. If ``use_enum_name`` is set to
``True``, then the value *must* be the key of the enum entry. If
``use_enum_name`` is set to ``False``, the value *must* be the value of the
enum entry. By default, this deserializer tries both.
:param obj: the serialized enum.
:param cls: the enum class.
:param use_enum_name: determines whether the name or the value of an enum
element should be used.
:param kwargs: not used.
:return: the corresponding enum element instance.
"""
if use_enum_name:
result = cls[obj]
elif use_enum_name is False:
result = cls(obj)
else: # use_enum_name is None
try:
result = cls[obj]
except KeyError:
result = cls(obj) # May raise a ValueError (which is expected).
return result

View File

@@ -0,0 +1,30 @@
from collections.abc import Mapping, Iterable
from typing import Iterable as IterableType
from jsons._compatibility_impl import get_naked_class
from jsons.deserializers.default_list import default_list_deserializer
def default_iterable_deserializer(
obj: list,
cls: type,
**kwargs) -> Iterable:
"""
Deserialize a (JSON) list into an ``Iterable`` by deserializing all items
of that list. The given obj is assumed to be homogeneous; if the list has a
generic type (e.g. Set[datetime]) then it is assumed that all elements can
be deserialized to that type.
:param obj: The list that needs deserializing to an ``Iterable``.
:param cls: The type, optionally with a generic (e.g. Deque[str]).
:param kwargs: Any keyword arguments.
:return: A deserialized ``Iterable`` (e.g. ``set``) instance.
"""
cls_ = Mapping
if hasattr(cls, '__args__'):
cls_ = IterableType[cls.__args__]
list_ = default_list_deserializer(obj, cls_, **kwargs)
result = list_
naked_cls = get_naked_class(cls)
if not isinstance(result, naked_cls):
result = naked_cls(list_)
return result

View File

@@ -0,0 +1,69 @@
from multiprocessing import Process
from typing import Type
from typish import get_args
from jsons._common_impl import StateHolder
from jsons._load_impl import load
from jsons._multitasking import multi_task
from jsons.exceptions import JsonsError, DeserializationError
def default_list_deserializer(
obj: list,
cls: type = None,
*,
warn_on_fail: bool = False,
tasks: int = 1,
task_type: type = Process,
fork_inst: Type[StateHolder] = StateHolder,
**kwargs) -> list:
"""
Deserialize a list by deserializing all items of that list.
:param obj: the list that needs deserializing.
:param cls: the type optionally with a generic (e.g. List[str]).
:param warn_on_fail: if ``True``, will warn upon any failure and continue.
:param tasks: the allowed number of tasks (threads or processes).
:param task_type: the type that is used for multitasking.
:param fork_inst: if given, it uses this fork of ``JsonSerializable``.
: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 = _do_load(obj, cls_, warn_on_fail, fork_inst, kwargs_)
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(
obj: list,
cls: type,
warn_on_fail: bool,
fork_inst: Type[StateHolder],
kwargs) -> list:
result = []
for index, elem in enumerate(obj):
try:
result.append(load(elem, cls=cls, tasks=1, fork_inst=fork_inst, **kwargs))
except DeserializationError as err:
new_msg = ('Could not deserialize element at index %s. %s' %
(index, err.message))
if warn_on_fail:
fork_inst._warn(new_msg, 'element-not-deserialized')
else:
new_err = DeserializationError(new_msg, err.source, err.target)
raise new_err from err
return result

View File

@@ -0,0 +1,27 @@
from collections.abc import Mapping
from typing import Mapping as MappingType
from typish import get_args, get_origin
from jsons.deserializers.default_dict import default_dict_deserializer
def default_mapping_deserializer(obj: dict, cls: type, **kwargs) -> Mapping:
"""
Deserialize a (JSON) dict into a mapping by deserializing all items of that
dict.
:param obj: the dict that needs deserializing.
:param cls: the type, optionally with a generic (e.g. Set[str]).
:param kwargs: any keyword arguments.
:return: a deserialized set instance.
"""
cls_ = Mapping
cls_args = get_args(cls)
if cls_args:
cls_ = MappingType[cls_args]
dict_ = default_dict_deserializer(obj, cls_, **kwargs)
result = dict_
# Strip any generics from cls to allow for an instance check.
if not isinstance(result, get_origin(cls)):
result = cls(dict_)
return result

View File

@@ -0,0 +1,19 @@
from typing import Optional
from jsons.exceptions import DeserializationError
def default_nonetype_deserializer(obj: object,
cls: Optional[type] = None,
**kwargs) -> object:
"""
Deserialize a ``NoneType``.
:param obj: the value that is to be deserialized.
:param cls: not used.
:param kwargs: not used.
:return: ``obj``.
"""
if obj is not None:
raise DeserializationError('Cannot deserialize {} as NoneType'
.format(obj), source=obj, target=cls)
return obj

View File

@@ -0,0 +1,200 @@
import inspect
from typing import Optional, Callable, Tuple
from jsons._cache import cached
from jsons._common_impl import (
get_class_name,
META_ATTR,
get_cls_from_str,
determine_precedence,
can_match_with_none
)
from jsons._compatibility_impl import get_type_hints
from jsons._load_impl import load
from jsons.exceptions import SignatureMismatchError, UnfulfilledArgumentError
def default_object_deserializer(
obj: dict,
cls: type,
*,
key_transformer: Optional[Callable[[str], str]] = None,
strict: bool = False,
**kwargs) -> object:
"""
Deserialize ``obj`` into an instance of type ``cls``. If ``obj`` contains
keys with a certain case style (e.g. camelCase) that do not match the style
of ``cls`` (e.g. snake_case), a key_transformer should be used (e.g.
KEY_TRANSFORMER_SNAKECASE).
:param obj: a serialized instance of ``cls``.
:param cls: the type to which ``obj`` should be deserialized.
:param key_transformer: a function that transforms the keys in order to
match the attribute names of ``cls``.
:param strict: deserialize in strict mode.
:param kwargs: any keyword arguments that may be passed to the
deserializers.
:return: an instance of type ``cls``.
"""
obj, kwargs = _check_and_transform_keys(obj, key_transformer, **kwargs)
kwargs['strict'] = strict
constructor_args = _get_constructor_args(obj, cls, **kwargs)
remaining_attrs = _get_remaining_args(obj, cls, constructor_args,
strict, kwargs['fork_inst'])
instance = cls(**constructor_args)
_set_remaining_attrs(instance, remaining_attrs, **kwargs)
return instance
def _get_constructor_args(
obj,
cls,
meta_hints,
attr_getters=None,
**kwargs) -> dict:
# Loop through the signature of cls: the type we try to deserialize to. For
# every required parameter, we try to get the corresponding value from
# json_obj.
signature_parameters = _get_signature(cls)
hints = get_type_hints(cls.__init__, fallback_ns=cls.__module__)
attr_getters = dict(**(attr_getters or {}))
result = {}
for sig_key, sig in signature_parameters.items():
if sig_key != 'self':
key, value = _get_value_for_attr(obj=obj,
orig_cls=cls,
meta_hints=meta_hints,
attr_getters=attr_getters,
sig_key=sig_key,
cls=hints.get(sig_key, None),
sig=sig,
**kwargs)
if key:
result[key] = value
return result
@cached
def _get_signature(cls):
return inspect.signature(cls.__init__).parameters
def _get_value_for_attr(
obj,
cls,
orig_cls,
sig_key,
sig,
meta_hints,
attr_getters,
**kwargs):
# Find a value for the attribute (with signature sig_key).
if obj and sig_key in obj:
# This argument is in obj.
result = sig_key, _get_value_from_obj(obj, cls, sig, sig_key,
meta_hints, **kwargs)
elif sig_key in attr_getters:
# There exists an attr_getter for this argument.
attr_getter = attr_getters.pop(sig_key)
result = sig_key, attr_getter()
elif sig.default != inspect.Parameter.empty:
# There is a default value for this argument.
result = sig_key, sig.default
elif sig.kind in (inspect.Parameter.VAR_POSITIONAL,
inspect.Parameter.VAR_KEYWORD):
# This argument is either *args or **kwargs.
result = None, None
elif can_match_with_none(cls):
# It is fine that there is no value.
result = sig_key, None
else:
raise UnfulfilledArgumentError(
'No value found for "{}".'.format(sig_key), sig_key, obj, orig_cls)
return result
def _remove_prefix(prefix: str, s: str) -> str:
if s.startswith(prefix):
return s[len(prefix):] or '/' # Special case: map the empty string to '/'
return s
def _get_value_from_obj(obj, cls, sig, sig_key, meta_hints, **kwargs):
# Obtain the value for the attribute with the given signature from the
# given obj. Try to obtain the class of this attribute from the meta info
# or from type hints.
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 = {
_remove_prefix(cls_key, key): meta_hints[key]
for key in meta_hints
}
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
def _set_remaining_attrs(instance,
remaining_attrs,
attr_getters,
**kwargs):
# Set any remaining attributes on the newly created instance.
attr_getters = attr_getters or {}
for attr_name in remaining_attrs:
annotations = get_type_hints(instance.__class__)
attr_type = annotations.get(attr_name)
if isinstance(remaining_attrs[attr_name], dict) \
and '-keys' in remaining_attrs[attr_name] \
and not attr_type:
fork_inst = kwargs['fork_inst']
fork_inst._warn('A dict with -keys was detected without a type '
'hint for attribute `{}`. This probably means '
'that you did not provide an annotation in your '
'class (ending up in __annotations__).'
.format(attr_name), 'hashed-keys-without-hint')
attr_type = attr_type or type(remaining_attrs[attr_name])
loaded_attr = load(remaining_attrs[attr_name], attr_type, **kwargs)
try:
setattr(instance, attr_name, loaded_attr)
except AttributeError:
pass # This is raised when a @property does not have a setter.
for attr_name, getter in attr_getters.items():
setattr(instance, attr_name, getter())
def _check_and_transform_keys(obj: dict,
key_transformer: Optional[Callable[[str], str]],
**kwargs) -> Tuple[dict, dict]:
if key_transformer:
obj = {key_transformer(key): obj[key] for key in obj}
kwargs = {
**kwargs,
'key_transformer': key_transformer
}
return obj, kwargs
def _get_remaining_args(obj: dict,
cls: type,
constructor_args: dict,
strict: bool,
fork_inst: type) -> dict:
# Get the remaining args or raise if strict and the signature is unmatched.
remaining_attrs = {attr_name: obj[attr_name] for attr_name in obj
if attr_name not in constructor_args
and attr_name != META_ATTR}
if strict and remaining_attrs:
unexpected_arg = list(remaining_attrs.keys())[0]
err_msg = ('Type "{}" does not expect "{}".'
.format(get_class_name(cls), unexpected_arg))
raise SignatureMismatchError(err_msg, unexpected_arg, obj, cls)
return remaining_attrs

View File

@@ -0,0 +1,12 @@
from pathlib import PurePath
def default_path_deserializer(obj: str, cls: type = PurePath, **kwargs) -> PurePath:
"""
Deserialize a string to a `pathlib.PurePath` object. Since ``pathlib``
implements ``PurePath``, no filename or existence checks are performed.
:param obj: the string to deserialize.
:param kwargs: not used.
:return: a ``str``.
"""
return cls(obj)

View File

@@ -0,0 +1,24 @@
from typing import Optional
from jsons.exceptions import DeserializationError
def default_primitive_deserializer(obj: object,
cls: Optional[type] = None,
**kwargs) -> object:
"""
Deserialize a primitive: it simply returns the given primitive.
:param obj: the value that is to be deserialized.
:param cls: not used.
:param kwargs: not used.
:return: ``obj``.
"""
result = obj
if obj is not None and not isinstance(obj, cls):
try:
result = cls(obj)
except ValueError as err:
raise DeserializationError(
'Could not cast "{}" into "{}"'.format(obj, cls.__name__),
obj, cls) from err
return result

View File

@@ -0,0 +1,27 @@
from datetime import datetime
from typing import Optional
from jsons._load_impl import load
from jsons.deserializers.default_primitive import default_primitive_deserializer
from jsons.exceptions import DeserializationError
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.
"""
target_is_str = cls is str and not kwargs.get('_inferred_cls')
if target_is_str:
return str(obj)
try:
result = load(obj, datetime, **kwargs)
except DeserializationError:
result = default_primitive_deserializer(obj, str)
return result

View File

@@ -0,0 +1,16 @@
from datetime import time
from jsons._datetime_impl import get_datetime_inst, RFC3339_TIME_PATTERN
def default_time_deserializer(obj: str,
cls: type = time,
**kwargs) -> time:
"""
Deserialize a string with an RFC3339 pattern to a time instance.
:param obj: the string that is to be deserialized.
:param cls: not used.
:param kwargs: not used.
:return: a ``datetime.time`` instance.
"""
return get_datetime_inst(obj, RFC3339_TIME_PATTERN).time()

View File

@@ -0,0 +1,14 @@
from datetime import timedelta
def default_timedelta_deserializer(obj: float,
cls: type = float,
**kwargs) -> timedelta:
"""
Deserialize a float to a timedelta instance.
:param obj: the float that is to be deserialized.
:param cls: not used.
:param kwargs: not used.
:return: a ``datetime.timedelta`` instance.
"""
return timedelta(seconds=obj)

View File

@@ -0,0 +1,16 @@
from datetime import timezone, timedelta
from jsons._load_impl import load
def default_timezone_deserializer(obj: dict,
cls: type = timezone,
**kwargs) -> timezone:
"""
Deserialize a dict to a timezone instance.
:param obj: the dict that is to be deserialized.
:param cls: not used.
:param kwargs: not used.
:return: a ``datetime.timezone`` instance.
"""
return timezone(load(obj['offset'], timedelta), obj['name'])

View File

@@ -0,0 +1,88 @@
from typing import Callable, Optional, Union
from typish import get_args
from jsons._common_impl import NoneType
from jsons._compatibility_impl import (get_type_hints, get_union_params,
tuple_with_ellipsis)
from jsons._load_impl import load
from jsons.exceptions import UnfulfilledArgumentError
def default_tuple_deserializer(obj: list,
cls: type = None,
*,
key_transformer: Optional[Callable[[str], str]] = None,
**kwargs) -> object:
"""
Deserialize a (JSON) list into a tuple by deserializing all items of that
list.
: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, key_transformer=key_transformer, **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 default_namedtuple_deserializer(
obj: Union[list, dict],
cls: type,
*,
key_transformer: Optional[Callable[[str], str]] = None,
**kwargs) -> object:
"""
Deserialize a (JSON) list or dict into a named tuple by deserializing all
items of that list/dict.
:param obj: the tuple that needs deserializing.
:param cls: the NamedTuple.
:param kwargs: any keyword arguments.
:return: a deserialized named tuple (i.e. an instance of a class).
"""
is_dict = isinstance(obj, dict)
key_tfr = key_transformer or (lambda key: key)
if is_dict:
tfm_obj = {key_tfr(k): v for k, v in obj.items()}
args = []
for index, field_name in enumerate(cls._fields):
if index < len(obj):
if is_dict:
field = tfm_obj[field_name]
else:
field = obj[index]
else:
field = cls._field_defaults.get(field_name, None)
# _field_types has been deprecated in favor of __annotations__ in Python 3.8
if hasattr(cls, '__annotations__'):
# It is important to use get_type_hints so that forward references get resolved,
# rather than access __annotations__ directly
field_types = get_type_hints(cls)
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_, key_transformer=key_transformer, **kwargs)
args.append(loaded_field)
inst = cls(*args)
return inst

View File

@@ -0,0 +1,30 @@
from typing import Union
from jsons._common_impl import get_class_name
from jsons._compatibility_impl import get_union_params
from jsons._load_impl import load
from jsons.exceptions import JsonsError, DeserializationError
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)

View File

@@ -0,0 +1,16 @@
from typing import Optional
from uuid import UUID
def default_uuid_deserializer(obj: str,
cls: Optional[type] = None,
**kwargs) -> UUID:
"""
Deserialize a UUID. Expected format for string is specified in RFC 4122.
e.g. '12345678-1234-1234-1234-123456789abc'
:param obj: the string that is to be deserialized.
:param cls: not used.
:param kwargs: any keyword arguments.
:return: the deserialized obj.
"""
return UUID(obj)

View File

@@ -0,0 +1,14 @@
try:
from zoneinfo import ZoneInfo
def default_zone_info_deserializer(obj: ZoneInfo, *_, **__) -> ZoneInfo:
"""
Deserialize a ZoneInfo.
:param obj: a serialized ZoneInfo object.
:return: an instance of ZoneInfo.
"""
return ZoneInfo(obj['key'])
except ImportError:
default_zone_info_deserializer = None

View File

@@ -0,0 +1,196 @@
"""
Contains the classes that may be raised by jsons.
"""
from json import JSONDecodeError
from typing import Optional
class JsonsError(Exception):
"""
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.
"""
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 DecodeError(DeserializationError, JSONDecodeError):
"""
Raised when decoding a string or bytes to Python types failed. This error
is actually a wrapper around `json.JSONDecodeError`.
"""
def __init__(self, message: str, source: object, target: type,
error: JSONDecodeError):
"""
Constructor.
:param message: the message of this error.
:param source: the object that was to be deserialized.
:param target: the type to which `source` was to be deserialized.
:param error: the wrapped `JSONDecodeError`.
"""
DeserializationError.__init__(self, message, source, target)
JSONDecodeError.__init__(self, message, error.doc, error.pos)
class UnfulfilledArgumentError(DeserializationError, ArgumentError):
"""
Raised on a deserialization failure when an argument could not be fulfilled
by the given object attr_getter.
"""
def __init__(self,
message: str,
argument: str,
source: object,
target: type):
"""
Constructor.
:param message: the message of this error.
:param argument: the argument that was unfulfilled.
:param source: the object that was to be deserialized.
:param target: the type to which `source` was to be deserialized.
"""
DeserializationError.__init__(self, message, source, target)
ArgumentError.__init__(self, message, argument)
class SignatureMismatchError(DeserializationError, ArgumentError):
"""
Raised when the source could not be deserialized into the target type due
to a mismatch between the source's attributes and the target's accepted
parameters. This error is raised in "strict-mode" only.
"""
def __init__(self,
message: str,
argument: str,
source: object,
target: type):
"""
Constructor.
:param message: the message of this error.
:param argument: the argument that caused the problem.
:param source: the object that was to be deserialized.
:param target: the type to which `source` was to be deserialized.
"""
DeserializationError.__init__(self, message, source, target)
ArgumentError.__init__(self, message, argument)
class UnknownClassError(DeserializationError):
"""
Raised when jsons failed to find a type instance to deserialize to. If this
error occurs, consider using ``jsons.announce_class``.
"""
def __init__(self, message: str, source: object, target_name: str):
"""
Constructor.
:param message: the message of this error.
:param source: the object that was to be deserialized.
: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)

View File

@@ -0,0 +1,6 @@
"""
This package contains default serializers. You can override the
serialization process of a particular type as follows:
``jsons.set_serializer(custom_serializer, SomeClass)``
"""

Some files were not shown because too many files have changed in this diff Show More