84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import typing as t
|
||
|
|
import zlib
|
||
|
|
|
||
|
|
from ._json import _CompactJSON
|
||
|
|
from .encoding import base64_decode
|
||
|
|
from .encoding import base64_encode
|
||
|
|
from .exc import BadPayload
|
||
|
|
from .serializer import _PDataSerializer
|
||
|
|
from .serializer import Serializer
|
||
|
|
from .timed import TimedSerializer
|
||
|
|
|
||
|
|
|
||
|
|
class URLSafeSerializerMixin(Serializer[str]):
|
||
|
|
"""Mixed in with a regular serializer it will attempt to zlib
|
||
|
|
compress the string to make it shorter if necessary. It will also
|
||
|
|
base64 encode the string so that it can safely be placed in a URL.
|
||
|
|
"""
|
||
|
|
|
||
|
|
default_serializer: _PDataSerializer[str] = _CompactJSON
|
||
|
|
|
||
|
|
def load_payload(
|
||
|
|
self,
|
||
|
|
payload: bytes,
|
||
|
|
*args: t.Any,
|
||
|
|
serializer: t.Any | None = None,
|
||
|
|
**kwargs: t.Any,
|
||
|
|
) -> t.Any:
|
||
|
|
decompress = False
|
||
|
|
|
||
|
|
if payload.startswith(b"."):
|
||
|
|
payload = payload[1:]
|
||
|
|
decompress = True
|
||
|
|
|
||
|
|
try:
|
||
|
|
json = base64_decode(payload)
|
||
|
|
except Exception as e:
|
||
|
|
raise BadPayload(
|
||
|
|
"Could not base64 decode the payload because of an exception",
|
||
|
|
original_error=e,
|
||
|
|
) from e
|
||
|
|
|
||
|
|
if decompress:
|
||
|
|
try:
|
||
|
|
json = zlib.decompress(json)
|
||
|
|
except Exception as e:
|
||
|
|
raise BadPayload(
|
||
|
|
"Could not zlib decompress the payload before decoding the payload",
|
||
|
|
original_error=e,
|
||
|
|
) from e
|
||
|
|
|
||
|
|
return super().load_payload(json, *args, **kwargs)
|
||
|
|
|
||
|
|
def dump_payload(self, obj: t.Any) -> bytes:
|
||
|
|
json = super().dump_payload(obj)
|
||
|
|
is_compressed = False
|
||
|
|
compressed = zlib.compress(json)
|
||
|
|
|
||
|
|
if len(compressed) < (len(json) - 1):
|
||
|
|
json = compressed
|
||
|
|
is_compressed = True
|
||
|
|
|
||
|
|
base64d = base64_encode(json)
|
||
|
|
|
||
|
|
if is_compressed:
|
||
|
|
base64d = b"." + base64d
|
||
|
|
|
||
|
|
return base64d
|
||
|
|
|
||
|
|
|
||
|
|
class URLSafeSerializer(URLSafeSerializerMixin, Serializer[str]):
|
||
|
|
"""Works like :class:`.Serializer` but dumps and loads into a URL
|
||
|
|
safe string consisting of the upper and lowercase character of the
|
||
|
|
alphabet as well as ``'_'``, ``'-'`` and ``'.'``.
|
||
|
|
"""
|
||
|
|
|
||
|
|
|
||
|
|
class URLSafeTimedSerializer(URLSafeSerializerMixin, TimedSerializer[str]):
|
||
|
|
"""Works like :class:`.TimedSerializer` but dumps and loads into a
|
||
|
|
URL safe string consisting of the upper and lowercase character of
|
||
|
|
the alphabet as well as ``'_'``, ``'-'`` and ``'.'``.
|
||
|
|
"""
|