platypush/platypush/schemas/__init__.py

from datetime import datetime, date
from enum import Enum
from typing import Any, Optional, Type, Union

from dateutil.parser import isoparse
from dateutil.tz import tzutc
from marshmallow import fields


class StrippedString(fields.Function):  # lgtm [py/missing-call-to-init]
    def __init__(self, *args, **kwargs):
        kwargs['serialize'] = self._strip
        kwargs['deserialize'] = self._strip
        super().__init__(*args, **kwargs)

    def _serialize(self, value, attr, obj, **kwargs) -> Optional[str]:
        if obj.get(attr) is not None:
            return self._strip(obj.get(attr))

    @staticmethod
    def _strip(value: str):
        return value.strip()


class Function(fields.Function):  # lgtm [py/missing-call-to-init]
    def _get_attr(self, obj, attr: str, _recursive=True):
        if self.attribute and _recursive:
            return self._get_attr(obj, self.attribute, False)
        if hasattr(obj, attr):
            return getattr(obj, attr)
        elif hasattr(obj, 'get'):
            return obj.get(attr)


class DateTime(Function):  # lgtm [py/missing-call-to-init]
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.metadata = {
            'example': datetime.now(tz=tzutc()).isoformat(),
            **(self.metadata or {}),
        }

    def _serialize(self, value, attr, obj, **kwargs) -> Optional[str]:
        value = normalize_datetime(self._get_attr(obj, attr))
        if value:
            return value.isoformat()

    def _deserialize(self, value, attr, data, **kwargs) -> Optional[datetime]:
        return normalize_datetime(value)


class Date(Function):  # lgtm [py/missing-call-to-init]
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.metadata = {
            'example': date.today().isoformat(),
            **(self.metadata or {}),
        }

    def _serialize(self, value, attr, obj, **kwargs) -> Optional[str]:
        value = normalize_datetime(self._get_attr(obj, attr))
        if value:
            return date(value.year, value.month, value.day).isoformat()

    def _deserialize(self, value, attr, data, **kwargs) -> Optional[date]:
        dt = normalize_datetime(value)
        return date.fromtimestamp(dt.timestamp())


class EnumField(Function):
    """
    Field that maps enum values.
    """

    # pylint: disable=redefined-builtin
    def __init__(self, *args, type: Type[Enum], **kwargs):
        super().__init__(*args, **kwargs)
        self.type = type

    def _serialize(self, value: Optional[Enum], *_, **__) -> Optional[Any]:
        return value.value if value is not None else None

    def _deserialize(self, value: Optional[Any], *_, **__) -> Optional[Any]:
        return self.type(value) if value is not None else None


def normalize_datetime(
    dt: Optional[Union[str, date, datetime]]
) -> Optional[Union[date, datetime]]:
    if not dt:
        return
    if isinstance(dt, (datetime, date)):
        return dt

    try:
        dt = float(dt)
        return datetime.fromtimestamp(dt)
    except (TypeError, ValueError):
        return isoparse(dt)