Source code for pycroft.helpers.i18n.serde

#  Copyright (c) 2022. The Pycroft Authors. See the AUTHORS file.
#  This file is part of the Pycroft project and licensed under the terms of
#  the Apache License, Version 2.0. See the LICENSE file for details
"""
pycroft.helpers.i18n.serde
~~~~~~~~~~~~~~~~~~~~~~~~~~
"""
from __future__ import annotations

import operator
import typing
from datetime import date, datetime, time, timedelta
from decimal import Decimal

from ..functional import identity
from .types import Interval, NegativeInfinity, PositiveInfinity, Bound, Money
from .utils import qualified_typename




def deserialize_money(v):
    try:
        return Money(Decimal(v[0]), v[1])
    except IndexError as e:
        raise ValueError from e





def serialize_interval(interval: Interval) -> dict[str, typing.Any]:
    lower = interval.lower_bound
    upper = interval.upper_bound
    lower_value = serialize_param(lower.value) if not lower.unbounded else None
    upper_value = serialize_param(upper.value) if not upper.unbounded else None
    return {
        "lower_closed": lower.closed,
        "lower_value": lower_value,
        "upper_closed": upper.closed,
        "upper_value": upper_value,
    }



# pycharm does not support building this dynamically.
# Thus, we need to hard-code this type alias.
Serializable = typing.Union[
    None,
    bool,
    str,
    int,
    float,
    Decimal,
    Money,
    date,
    datetime,
    time,
    timedelta,
    Interval,
]


# TODO be more specific about return type
serialize_map: dict[type, typing.Callable] = {
    type(None): identity,
    bool: identity,
    float: identity,
    str: identity,
    int: identity,
    Decimal: str,
    Money: lambda m: (str(m.value), m.currency),
    date: operator.methodcaller("isoformat"),
    datetime: operator.methodcaller("isoformat"),
    time: operator.methodcaller("isoformat"),
    timedelta: lambda v: {
        "days": v.days,
        "seconds": v.seconds,
        "microseconds": v.microseconds,
    },
    Interval: serialize_interval,
}




def deserialize_interval(value: dict[str, typing.Any]) -> Interval:
    try:
        lower_value = (
            deserialize_param(u)
            if (u := value["lower_value"]) is not None
            else NegativeInfinity
        )
        lower_closed = value["lower_closed"]
        upper_value = (
            deserialize_param(u)
            if (u := value["upper_value"]) is not None
            else PositiveInfinity
        )
        upper_closed = value["upper_closed"]
    except KeyError as e:
        raise ValueError("Could not deserialized from interval (missing key)") from e

    if not isinstance(lower_closed, bool):
        raise ValueError(
            "Could not deserialize to interval: "
            f"expected ['lower_closed'] to be bool, got {type(lower_closed)}"
        )
    if not isinstance(upper_closed, bool):
        raise ValueError(
            "Could not deserialize to interval: "
            f"expected ['upper_closed'] to be bool, got {type(upper_closed)}"
        )
    return Interval(
        lower_bound=Bound(lower_value, lower_closed),
        upper_bound=Bound(upper_value, upper_closed),
    )



_deserialize_type_map: dict[type, typing.Callable] = {
    type(None): identity,
    bool: identity,
    str: identity,
    int: identity,
    float: identity,
    Decimal: Decimal,
    Money: deserialize_money,
    date: date.fromisoformat,
    datetime: datetime.fromisoformat,
    time: time.fromisoformat,
    timedelta: lambda v: timedelta(**v),
    Interval: deserialize_interval,
}


# TODO be more specific
deserialize_map: dict[str, typing.Callable] = {
    qualified_typename(t): f for t, f in _deserialize_type_map.items()
}




def serialize_param(param):
    concrete_type = type(param)
    serializers = (
        (type_, serialize_map[type_])
        for type_ in concrete_type.__mro__
        if type_ in serialize_map
    )
    try:
        type_, serializer = next(serializers)
    except StopIteration:
        raise TypeError(
            "No serialization available for type {} or any"
            "supertype".format(qualified_typename(concrete_type))
        ) from None
    return {"type": qualified_typename(type_), "value": serializer(param)}





def deserialize_param(param):
    type_name = param["type"]
    try:
        deserializer = deserialize_map[type_name]
    except KeyError:
        raise TypeError(f"No deserialization available for type {type_name}") from None
    return deserializer(param["value"])