Skip to content

coder

cachetoolz.coder

Module interface.

cachetoolz.coder.Coder

Bases: CoderABC

Coder class.

Source code in cachetoolz/coder/__init__.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class Coder(CoderABC):
    """Coder class."""

    def encode(self, value: Any) -> str:
        """
        Encode value.

        Parameters
        ----------
        value : Any
            Value to encode.

        Returns
        -------
        str
            Value encoded.
        """
        return json.dumps(value, cls=Encoder)

    def decode(self, value: str) -> Dict[str, Any]:
        """
        Decode value.

        Parameters
        ----------
        value : Any
            Value to decode.

        Returns
        -------
        Dict[str, Any]
            Value decoded.
        """
        return json.loads(value, cls=Decoder)

    @staticmethod
    def register(serializer: Union[Type[SerializerABC], SerializerABC]):
        """
        Register a JSON serializer class.

        You can register a class for decoding, it needs to have the ``encode``
        and ``decode`` methods where the ``encode`` method must have a
        parameter called ``value`` and must have the type annotated.
        These methods can be ``instance``, ``@staticmethod``, or
        ``@classmethod``.
        The decode function will receive the exact value that is returned by
        the encode function.

        Parameters
        ----------
        serializer : Type[SerializerABC] | SerializerABC
            Serializer class.

        Returns
        -------
        Type[SerializerABC] | SerializerABC
            Serializer class.

        Examples
        --------
        Class methods

        >>> from collections import deque
        >>> @coder.register
        >>> class DequeSerializer:
        ...     @classmethod
        ...     def encode(cls, value: deque):
        ...         return {'iterable': list(value), 'maxlen': value.maxlen}
        ...
        ...     @classmethod
        ...     def decode(cls, value):
        ...         return deque(val['iterable'], val['maxlen'])
        ...

        Static methods

        >>> from collections import deque
        >>> @coder.register
        >>> class DequeSerializer:
        ...     @staticmethod
        ...     def encode(value: deque):
        ...         return {'iterable': list(value), 'maxlen': value.maxlen}
        ...
        ...     @staticmethod
        ...     def decode(value):
        ...         return deque(val['iterable'], val['maxlen'])
        ...

        Instace methods

        >>> from collections import deque
        >>> @coder.register
        >>> class DequeSerializer:
        ...     def encode(self, value: deque):
        ...         return {'iterable': list(value), 'maxlen': value.maxlen}
        ...
        ...     def decode(self, value):
        ...         return deque(val['iterable'], val['maxlen'])
        ...

        When registering a class, it will be instantiated.
        Therefore, if the class requires any initialization parameters,
        you can register an instance of it along with the necessary parameters.

        >>> from collections import deque
        >>> class DequeCoder:
        ...     def __init__(self, foo):
        ...         self.foo = foo
        ...
        ...      def encode(self, value: deque):
        ...         return {'iterable': list(value), 'maxlen': value.maxlen}
        ...
        ...     def decode(self, value):
        ...         return deque(val['iterable'], val['maxlen'])
        ...
        >>> coder.register(DequeCoder(foo='bar'))
        """
        if not hasattr(serializer, 'encode') or not hasattr(
            serializer, 'decode'
        ):
            raise RegistryError(
                'Sereliazador is not valid, the class must have implemented '
                'the `encode(value: type)` and `decode(value)` methods'
            )

        instance = serializer() if isclass(serializer) else serializer

        name = decoder_name(instance).replace('serializer', '').strip('_')

        encoder_register(name)(instance.encode)
        decoder_register(name)(instance.decode)

        return serializer

encode(value: Any) -> str

Encode value.

Parameters:

Name Type Description Default
value Any

Value to encode.

 required

Returns:

Type Description
str

Value encoded.
Source code in cachetoolz/coder/__init__.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def encode(self, value: Any) -> str:
    """
    Encode value.

    Parameters
    ----------
    value : Any
        Value to encode.

    Returns
    -------
    str
        Value encoded.
    """
    return json.dumps(value, cls=Encoder)

decode(value: str) -> Dict[str, Any]

Decode value.

Parameters:

Name Type Description Default
value Any

Value to decode.

 required

Returns:

Type Description
Dict[str, Any]

Value decoded.
Source code in cachetoolz/coder/__init__.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def decode(self, value: str) -> Dict[str, Any]:
    """
    Decode value.

    Parameters
    ----------
    value : Any
        Value to decode.

    Returns
    -------
    Dict[str, Any]
        Value decoded.
    """
    return json.loads(value, cls=Decoder)

register(serializer: Union[Type[SerializerABC], SerializerABC]) staticmethod

Register a JSON serializer class.

You can register a class for decoding, it needs to have the encode and decode methods where the encode method must have a parameter called value and must have the type annotated. These methods can be instance, @staticmethod, or @classmethod. The decode function will receive the exact value that is returned by the encode function.

Parameters:

Name Type Description Default
serializer Type[SerializerABC] | SerializerABC

Serializer class.

 required

Returns:

Type Description
Type[SerializerABC] | SerializerABC

Serializer class.

Examples:

Class methods

>>> from collections import deque
>>> @coder.register
>>> class DequeSerializer:
...     @classmethod
...     def encode(cls, value: deque):
...         return {'iterable': list(value), 'maxlen': value.maxlen}
...
...     @classmethod
...     def decode(cls, value):
...         return deque(val['iterable'], val['maxlen'])
...

Static methods

>>> from collections import deque
>>> @coder.register
>>> class DequeSerializer:
...     @staticmethod
...     def encode(value: deque):
...         return {'iterable': list(value), 'maxlen': value.maxlen}
...
...     @staticmethod
...     def decode(value):
...         return deque(val['iterable'], val['maxlen'])
...

Instace methods

>>> from collections import deque
>>> @coder.register
>>> class DequeSerializer:
...     def encode(self, value: deque):
...         return {'iterable': list(value), 'maxlen': value.maxlen}
...
...     def decode(self, value):
...         return deque(val['iterable'], val['maxlen'])
...

When registering a class, it will be instantiated. Therefore, if the class requires any initialization parameters, you can register an instance of it along with the necessary parameters.

>>> from collections import deque
>>> class DequeCoder:
...     def __init__(self, foo):
...         self.foo = foo
...
...      def encode(self, value: deque):
...         return {'iterable': list(value), 'maxlen': value.maxlen}
...
...     def decode(self, value):
...         return deque(val['iterable'], val['maxlen'])
...
>>> coder.register(DequeCoder(foo='bar'))
Source code in cachetoolz/coder/__init__.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@staticmethod
def register(serializer: Union[Type[SerializerABC], SerializerABC]):
    """
    Register a JSON serializer class.

    You can register a class for decoding, it needs to have the ``encode``
    and ``decode`` methods where the ``encode`` method must have a
    parameter called ``value`` and must have the type annotated.
    These methods can be ``instance``, ``@staticmethod``, or
    ``@classmethod``.
    The decode function will receive the exact value that is returned by
    the encode function.

    Parameters
    ----------
    serializer : Type[SerializerABC] | SerializerABC
        Serializer class.

    Returns
    -------
    Type[SerializerABC] | SerializerABC
        Serializer class.

    Examples
    --------
    Class methods

    >>> from collections import deque
    >>> @coder.register
    >>> class DequeSerializer:
    ...     @classmethod
    ...     def encode(cls, value: deque):
    ...         return {'iterable': list(value), 'maxlen': value.maxlen}
    ...
    ...     @classmethod
    ...     def decode(cls, value):
    ...         return deque(val['iterable'], val['maxlen'])
    ...

    Static methods

    >>> from collections import deque
    >>> @coder.register
    >>> class DequeSerializer:
    ...     @staticmethod
    ...     def encode(value: deque):
    ...         return {'iterable': list(value), 'maxlen': value.maxlen}
    ...
    ...     @staticmethod
    ...     def decode(value):
    ...         return deque(val['iterable'], val['maxlen'])
    ...

    Instace methods

    >>> from collections import deque
    >>> @coder.register
    >>> class DequeSerializer:
    ...     def encode(self, value: deque):
    ...         return {'iterable': list(value), 'maxlen': value.maxlen}
    ...
    ...     def decode(self, value):
    ...         return deque(val['iterable'], val['maxlen'])
    ...

    When registering a class, it will be instantiated.
    Therefore, if the class requires any initialization parameters,
    you can register an instance of it along with the necessary parameters.

    >>> from collections import deque
    >>> class DequeCoder:
    ...     def __init__(self, foo):
    ...         self.foo = foo
    ...
    ...      def encode(self, value: deque):
    ...         return {'iterable': list(value), 'maxlen': value.maxlen}
    ...
    ...     def decode(self, value):
    ...         return deque(val['iterable'], val['maxlen'])
    ...
    >>> coder.register(DequeCoder(foo='bar'))
    """
    if not hasattr(serializer, 'encode') or not hasattr(
        serializer, 'decode'
    ):
        raise RegistryError(
            'Sereliazador is not valid, the class must have implemented '
            'the `encode(value: type)` and `decode(value)` methods'
        )

    instance = serializer() if isclass(serializer) else serializer

    name = decoder_name(instance).replace('serializer', '').strip('_')

    encoder_register(name)(instance.encode)
    decoder_register(name)(instance.decode)

    return serializer

cachetoolz.coder.encoder

Encoder module.

cachetoolz.coder.encoder.Encoder

Bases: JSONEncoder

JSON encoder class.

Source code in cachetoolz/coder/encoder.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
class Encoder(JSONEncoder):
    """JSON encoder class."""

    def default(self, o: Any):
        """
        Turn a python object into a json object.

        Parameters
        ----------
        o : Any
            Python object.

        Returns
        -------
        Any
            Object encoded.
        """
        return encode(o)

cachetoolz.coder.encoder.register(name: str) -> types.Decorator

Register a encoder.

Parameters:

Name Type Description Default
name str

Encoder name.

 required

Returns:

Type Description
types.Decorator

Function Decorator.

Examples:

>>> from collections import deque
>>> @register('deque')
... def _(value: deque):
...     return {'iterable': list(value), 'maxlen': value.maxlen}
Source code in cachetoolz/coder/encoder.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def register(name: str) -> types.Decorator:
    """
    Register a encoder.

    Parameters
    ----------
    name : str
        Encoder name.

    Returns
    -------
    types.Decorator
        Function Decorator.

    Examples
    --------
    >>> from collections import deque
    >>> @register('deque')
    ... def _(value: deque):
    ...     return {'iterable': list(value), 'maxlen': value.maxlen}
    """

    def wrapper(func: types.Encoder) -> types.Encoder:
        if not callable(func):
            raise RegistryError('encoder needs to be a callable')
        if 'value' not in (annotations := get_annotations(func)):
            raise RegistryError(
                'encoder needs to have a parameter named `value` '
                'and it needs to have the type annotated'
            )

        _type = annotations['value']

        encode.register(_type)(
            lambda value: {'__val': func(value), '__decoder': name}
        )
        return func

    return wrapper

cachetoolz.coder.decoder

Decoder module.

cachetoolz.coder.decoder.Decoder

Bases: JSONDecoder

JSON decoder class.

Parameters:

Name Type Description Default
**kwargs Any

JSONDecoder parameters.

 {}
Source code in cachetoolz/coder/decoder.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class Decoder(JSONDecoder):
    """
    JSON decoder class.

    Parameters
    ----------
    **kwargs : Any
        JSONDecoder parameters.
    """

    DECODERS: ClassVar[Dict[str, types.Decoder]] = {
        'time': time.fromisoformat,
        'date': date.fromisoformat,
        'datetime': datetime.fromisoformat,
        'timedelta': lambda val: timedelta(seconds=val),
        'decimal': Decimal,
        'uuid': UUID,
        'posixpath': Path,
        'ipv4address': IPv4Address,
        'ipv4interface': IPv4Interface,
        'ipv4network': IPv4Network,
        'ipv6address': IPv6Address,
        'ipv6interface': IPv6Interface,
        'ipv6network': IPv6Network,
        'set': set,
        'frozenset': frozenset,
        'bytes': lambda val: val['bytes'].encode(val['encoding']),
        'deque': lambda val: deque(val['iterable'], val['maxlen']),
        'pattern': lambda val: re.compile(
            json.loads(val['pattern']), val['flags']
        ),
    }

    def __init__(self, **kwargs: Any):
        """
        Initialize the instance.

        Parameters
        ----------
        **kwargs : Any
            JSONDecoder parameters.
        """
        kwargs['object_hook'] = self._object_hook
        super().__init__(**kwargs)

    def _object_hook(self, obj) -> Any:
        if not (decoder := obj.get('__decoder')):
            return obj

        if decoder not in self.DECODERS:
            raise UnknownDecoderError(f'Decoder "{decoder}" not registered')

        return self.DECODERS[decoder](obj['__val'])

cachetoolz.coder.decoder.register(name: str) -> types.Decorator

Register a decoder.

Parameters:

Name Type Description Default
name str

Decoder name.

 required

Returns:

Type Description
types.Decorator

Function Decorator.

Examples:

>>> from collections import deque
>>> from typing import Any
>>> @register('deque')
... def _(val: dict[str, Any]):
...     return deque(val['iterable'], val['maxlen'])
Source code in cachetoolz/coder/decoder.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def register(name: str) -> types.Decorator:
    """
    Register a decoder.

    Parameters
    ----------
    name : str
        Decoder name.

    Returns
    -------
    types.Decorator
        Function Decorator.

    Examples
    --------
    >>> from collections import deque
    >>> from typing import Any
    >>> @register('deque')
    ... def _(val: dict[str, Any]):
    ...     return deque(val['iterable'], val['maxlen'])
    """

    def wrapper(func: types.Decoder) -> types.Decoder:
        if not callable(func):
            raise RegistryError('decoder needs to be a callable')

        Decoder.DECODERS[name] = func
        return func

    return wrapper