You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
313 lines
8.6 KiB
Python
313 lines
8.6 KiB
Python
"""
|
|
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)
|