jsons.serializers.default_iterable

Complexity

Total Complexity

9

Size/Duplication

Total Lines	73
Duplicated Lines	0 %

Importance

Changes

0

3 Functions

Rating	Name	Duplication	Size	Complexity
A	default_iterable_serializer()	0	37	3
A	_get_subclasses()	0	16	5
A	f()	0	4	1

from collections.abc import Iterable
from multiprocessing import Process
from typing import Tuple

from typish import get_args

from jsons._dump_impl import dump
from jsons._multitasking import multi_task
from jsons.exceptions import SerializationError


def default_iterable_serializer(
        obj: Iterable,
        cls: type = None,
        *,
        strict: bool = False,
        tasks: int = 1,
        task_type: type = Process,
        **kwargs) -> list:
    """
    Serialize the given ``obj`` to a list of serialized objects.
    :param obj: the iterable that is to be serialized.
    :param cls: the (subscripted) type of the iterable.
    :param strict: a bool to determine if the serializer should be strict
    (i.e. only dumping stuff that is known to ``cls``).
    :param tasks: the allowed number of tasks (threads or processes).
    :param task_type: the type that is used for multitasking.
    :param kwargs: any keyword arguments that may be given to the serialization
    process.
    :return: a list of which all elements are serialized.
    """
    # The meta kwarg store_cls is filtered out, because an iterable should have
    # its own -meta attribute.
    kwargs_ = {**kwargs, 'strict': strict}
    kwargs_.pop('_store_cls', None)
    if strict:
        subclasses = _get_subclasses(obj, cls)
    else:
        subclasses = _get_subclasses(obj, None)

    if tasks < 2:
        result = [dump(elem, cls=subclasses[i], **kwargs_)
                  for i, elem in enumerate(obj)]
    else:
        zipped_objs = list(zip(obj, subclasses))
        result = multi_task(f, zipped_objs, tasks, task_type, **kwargs_)

    return result


def _get_subclasses(obj: Iterable, cls: type = None) -> Tuple[type, ...]:
    subclasses = (None,) * len(obj)
    if cls:
        args = get_args(cls)
        if len(args) == 1:
            # E.g. List[int]
            subclasses = args * len(obj)
        elif len(args) > 1:
            # E.g. Tuple[int, str, str]
            subclasses = args
    if len(subclasses) != len(obj):
        msg = ('Not enough generic types ({}) in {}, expected {} to match '
               'the iterable of length {}'
               .format(len(subclasses), cls, len(obj), len(obj)))
        raise SerializationError(msg)
    return subclasses


def f(obj_cls_tuple: Tuple[object, type], *args, **kwargs):
    kwargs_ = {**kwargs}
    kwargs_['tasks'] = 1
    return dump(*obj_cls_tuple, *args, **kwargs_)