JezK
Edit File: tag.py
""" Tagged JSON ~~~~~~~~~~~ A compact representation for lossless serialization of non-standard JSON types. :class:`~flask.sessions.SecureCookieSessionInterface` uses this to serialize the session data, but it may be useful in other places. It can be extended to support other types. .. autoclass:: TaggedJSONSerializer :members: .. autoclass:: JSONTag :members: Let's see an example that adds support for :class:`~collections.OrderedDict`. Dicts don't have an order in JSON, so to handle this we will dump the items as a list of ``[key, value]`` pairs. Subclass :class:`JSONTag` and give it the new key ``' od'`` to identify the type. The session serializer processes dicts first, so insert the new tag at the front of the order since ``OrderedDict`` must be processed before ``dict``. .. code-block:: python from flask.json.tag import JSONTag class TagOrderedDict(JSONTag): __slots__ = ('serializer',) key = ' od' def check(self, value): return isinstance(value, OrderedDict) def to_json(self, value): return [[k, self.serializer.tag(v)] for k, v in iteritems(value)] def to_python(self, value): return OrderedDict(value) app.session_interface.serializer.register(TagOrderedDict, index=0) """ import typing as t from base64 import b64decode from base64 import b64encode from datetime import datetime from uuid import UUID from markupsafe import Markup from werkzeug.http import http_date from werkzeug.http import parse_date from ..json import dumps from ..json import loads class JSONTag: """Base class for defining type tags for :class:`TaggedJSONSerializer`.""" __slots__ = ("serializer",) #: The tag to mark the serialized object with. If ``None``, this tag is #: only used as an intermediate step during tagging. key: t.Optional[str] = None def __init__(self, serializer: "TaggedJSONSerializer") -> None: """Create a tagger for the given serializer.""" self.serializer = serializer def check(self, value: t.Any) -> bool: """Check if the given value should be tagged by this tag.""" raise NotImplementedError def to_json(self, value: t.Any) -> t.Any: """Convert the Python object to an object that is a valid JSON type. The tag will be added later.""" raise NotImplementedError def to_python(self, value: t.Any) -> t.Any: """Convert the JSON representation back to the correct type. The tag will already be removed.""" raise NotImplementedError def tag(self, value: t.Any) -> t.Any: """Convert the value to a valid JSON type and add the tag structure around it.""" return {self.key: self.to_json(value)} class TagDict(JSONTag): """Tag for 1-item dicts whose only key matches a registered tag. Internally, the dict key is suffixed with `__`, and the suffix is removed when deserializing. """ __slots__ = () key = " di" def check(self, value: t.Any) -> bool: return ( isinstance(value, dict) and len(value) == 1 and next(iter(value)) in self.serializer.tags ) def to_json(self, value: t.Any) -> t.Any: key = next(iter(value)) return {f"{key}__": self.serializer.tag(value[key])} def to_python(self, value: t.Any) -> t.Any: key = next(iter(value)) return {key[:-2]: value[key]} class PassDict(JSONTag): __slots__ = () def check(self, value: t.Any) -> bool: return isinstance(value, dict) def to_json(self, value: t.Any) -> t.Any: # JSON objects may only have string keys, so don't bother tagging the # key here. return {k: self.serializer.tag(v) for k, v in value.items()} tag = to_json class TagTuple(JSONTag): __slots__ = () key = " t" def check(self, value: t.Any) -> bool: return isinstance(value, tuple) def to_json(self, value: t.Any) -> t.Any: return [self.serializer.tag(item) for item in value] def to_python(self, value: t.Any) -> t.Any: return tuple(value) class PassList(JSONTag): __slots__ = () def check(self, value: t.Any) -> bool: return isinstance(value, list) def to_json(self, value: t.Any) -> t.Any: return [self.serializer.tag(item) for item in value] tag = to_json class TagBytes(JSONTag): __slots__ = () key = " b" def check(self, value: t.Any) -> bool: return isinstance(value, bytes) def to_json(self, value: t.Any) -> t.Any: return b64encode(value).decode("ascii") def to_python(self, value: t.Any) -> t.Any: return b64decode(value) class TagMarkup(JSONTag): """Serialize anything matching the :class:`~markupsafe.Markup` API by having a ``__html__`` method to the result of that method. Always deserializes to an instance of :class:`~markupsafe.Markup`.""" __slots__ = () key = " m" def check(self, value: t.Any) -> bool: return callable(getattr(value, "__html__", None)) def to_json(self, value: t.Any) -> t.Any: return str(value.__html__()) def to_python(self, value: t.Any) -> t.Any: return Markup(value) class TagUUID(JSONTag): __slots__ = () key = " u" def check(self, value: t.Any) -> bool: return isinstance(value, UUID) def to_json(self, value: t.Any) -> t.Any: return value.hex def to_python(self, value: t.Any) -> t.Any: return UUID(value) class TagDateTime(JSONTag): __slots__ = () key = " d" def check(self, value: t.Any) -> bool: return isinstance(value, datetime) def to_json(self, value: t.Any) -> t.Any: return http_date(value) def to_python(self, value: t.Any) -> t.Any: return parse_date(value) class TaggedJSONSerializer: """Serializer that uses a tag system to compactly represent objects that are not JSON types. Passed as the intermediate serializer to :class:`itsdangerous.Serializer`. The following extra types are supported: * :class:`dict` * :class:`tuple` * :class:`bytes` * :class:`~markupsafe.Markup` * :class:`~uuid.UUID` * :class:`~datetime.datetime` """ __slots__ = ("tags", "order") #: Tag classes to bind when creating the serializer. Other tags can be #: added later using :meth:`~register`. default_tags = [ TagDict, PassDict, TagTuple, PassList, TagBytes, TagMarkup, TagUUID, TagDateTime, ] def __init__(self) -> None: self.tags: t.Dict[str, JSONTag] = {} self.order: t.List[JSONTag] = [] for cls in self.default_tags: self.register(cls) def register( self, tag_class: t.Type[JSONTag], force: bool = False, index: t.Optional[int] = None, ) -> None: """Register a new tag with this serializer. :param tag_class: tag class to register. Will be instantiated with this serializer instance. :param force: overwrite an existing tag. If false (default), a :exc:`KeyError` is raised. :param index: index to insert the new tag in the tag order. Useful when the new tag is a special case of an existing tag. If ``None`` (default), the tag is appended to the end of the order. :raise KeyError: if the tag key is already registered and ``force`` is not true. """ tag = tag_class(self) key = tag.key if key is not None: if not force and key in self.tags: raise KeyError(f"Tag '{key}' is already registered.") self.tags[key] = tag if index is None: self.order.append(tag) else: self.order.insert(index, tag) def tag(self, value: t.Any) -> t.Dict[str, t.Any]: """Convert a value to a tagged representation if necessary.""" for tag in self.order: if tag.check(value): return tag.tag(value) return value def untag(self, value: t.Dict[str, t.Any]) -> t.Any: """Convert a tagged representation back to the original type.""" if len(value) != 1: return value key = next(iter(value)) if key not in self.tags: return value return self.tags[key].to_python(value[key]) def dumps(self, value: t.Any) -> str: """Tag the value and dump it to a compact JSON string.""" return dumps(self.tag(value), separators=(",", ":")) def loads(self, value: str) -> t.Any: """Load data from a JSON string and deserialized any tagged objects.""" return loads(value, object_hook=self.untag)