1  """
   2  Tagged JSON
   3  ~~~~~~~~~~~
   4  
   5  A compact representation for lossless serialization of non-standard JSON
   6  types. :class:`~flask.sessions.SecureCookieSessionInterface` uses this
   7  to serialize the session data, but it may be useful in other places. It
   8  can be extended to support other types.
   9  
  10  .. autoclass:: TaggedJSONSerializer
  11      :members:
  12  
  13  .. autoclass:: JSONTag
  14      :members:
  15  
  16  Let's see an example that adds support for
  17  :class:`~collections.OrderedDict`. Dicts don't have an order in JSON, so
  18  to handle this we will dump the items as a list of ``[key, value]``
  19  pairs. Subclass :class:`JSONTag` and give it the new key ``' od'`` to
  20  identify the type. The session serializer processes dicts first, so
  21  insert the new tag at the front of the order since ``OrderedDict`` must
  22  be processed before ``dict``.
  23  
  24  .. code-block:: python
  25  
  26      from flask.json.tag import JSONTag
  27  
  28      class TagOrderedDict(JSONTag):
  29          __slots__ = ('serializer',)
  30          key = ' od'
  31  
  32          def check(self, value):
  33              return isinstance(value, OrderedDict)
  34  
  35          def to_json(self, value):
  36              return [[k, self.serializer.tag(v)] for k, v in iteritems(value)]
  37  
  38          def to_python(self, value):
  39              return OrderedDict(value)
  40  
  41      app.session_interface.serializer.register(TagOrderedDict, index=0)
  42  """
  43  
  44  from __future__ import annotations
  45  
  46  import typing as t
  47  from base64 import b64decode
  48  from base64 import b64encode
  49  from datetime import datetime
  50  from uuid import UUID
  51  
  52  from markupsafe import Markup
  53  from werkzeug.http import http_date
  54  from werkzeug.http import parse_date
  55  
  56  from ..json import dumps
  57  from ..json import loads
  58  
  59  
  60  class JSONTag:
  61      """Base class for defining type tags for :class:`TaggedJSONSerializer`."""
  62  
  63      __slots__ = ("serializer",)
  64  
  65      #: The tag to mark the serialized object with. If empty, this tag is
  66      #: only used as an intermediate step during tagging.
  67      key: str = ""
  68  
  69      def __init__(self, serializer: TaggedJSONSerializer) -> None:
  70          """Create a tagger for the given serializer."""
  71          self.serializer = serializer
  72  
  73      def check(self, value: t.Any) -> bool:
  74          """Check if the given value should be tagged by this tag."""
  75          raise NotImplementedError
  76  
  77      def to_json(self, value: t.Any) -> t.Any:
  78          """Convert the Python object to an object that is a valid JSON type.
  79          The tag will be added later."""
  80          raise NotImplementedError
  81  
  82      def to_python(self, value: t.Any) -> t.Any:
  83          """Convert the JSON representation back to the correct type. The tag
  84          will already be removed."""
  85          raise NotImplementedError
  86  
  87      def tag(self, value: t.Any) -> dict[str, t.Any]:
  88          """Convert the value to a valid JSON type and add the tag structure
  89          around it."""
  90          return {self.key: self.to_json(value)}
  91  
  92  
  93  class TagDict(JSONTag):
  94      """Tag for 1-item dicts whose only key matches a registered tag.
  95  
  96      Internally, the dict key is suffixed with `__`, and the suffix is removed
  97      when deserializing.
  98      """
  99  
 100      __slots__ = ()
 101      key = " di"
 102  
 103      def check(self, value: t.Any) -> bool:
 104          return (
 105              isinstance(value, dict)
 106              and len(value) == 1
 107              and next(iter(value)) in self.serializer.tags
 108          )
 109  
 110      def to_json(self, value: t.Any) -> t.Any:
 111          key = next(iter(value))
 112          return {f"{key}__": self.serializer.tag(value[key])}
 113  
 114      def to_python(self, value: t.Any) -> t.Any:
 115          key = next(iter(value))
 116          return {key[:-2]: value[key]}
 117  
 118  
 119  class PassDict(JSONTag):
 120      __slots__ = ()
 121  
 122      def check(self, value: t.Any) -> bool:
 123          return isinstance(value, dict)
 124  
 125      def to_json(self, value: t.Any) -> t.Any:
 126          # JSON objects may only have string keys, so don't bother tagging the
 127          # key here.
 128          return {k: self.serializer.tag(v) for k, v in value.items()}
 129  
 130      tag = to_json
 131  
 132  
 133  class TagTuple(JSONTag):
 134      __slots__ = ()
 135      key = " t"
 136  
 137      def check(self, value: t.Any) -> bool:
 138          return isinstance(value, tuple)
 139  
 140      def to_json(self, value: t.Any) -> t.Any:
 141          return [self.serializer.tag(item) for item in value]
 142  
 143      def to_python(self, value: t.Any) -> t.Any:
 144          return tuple(value)
 145  
 146  
 147  class PassList(JSONTag):
 148      __slots__ = ()
 149  
 150      def check(self, value: t.Any) -> bool:
 151          return isinstance(value, list)
 152  
 153      def to_json(self, value: t.Any) -> t.Any:
 154          return [self.serializer.tag(item) for item in value]
 155  
 156      tag = to_json
 157  
 158  
 159  class TagBytes(JSONTag):
 160      __slots__ = ()
 161      key = " b"
 162  
 163      def check(self, value: t.Any) -> bool:
 164          return isinstance(value, bytes)
 165  
 166      def to_json(self, value: t.Any) -> t.Any:
 167          return b64encode(value).decode("ascii")
 168  
 169      def to_python(self, value: t.Any) -> t.Any:
 170          return b64decode(value)
 171  
 172  
 173  class TagMarkup(JSONTag):
 174      """Serialize anything matching the :class:`~markupsafe.Markup` API by
 175      having a ``__html__`` method to the result of that method. Always
 176      deserializes to an instance of :class:`~markupsafe.Markup`."""
 177  
 178      __slots__ = ()
 179      key = " m"
 180  
 181      def check(self, value: t.Any) -> bool:
 182          return callable(getattr(value, "__html__", None))
 183  
 184      def to_json(self, value: t.Any) -> t.Any:
 185          return str(value.__html__())
 186  
 187      def to_python(self, value: t.Any) -> t.Any:
 188          return Markup(value)
 189  
 190  
 191  class TagUUID(JSONTag):
 192      __slots__ = ()
 193      key = " u"
 194  
 195      def check(self, value: t.Any) -> bool:
 196          return isinstance(value, UUID)
 197  
 198      def to_json(self, value: t.Any) -> t.Any:
 199          return value.hex
 200  
 201      def to_python(self, value: t.Any) -> t.Any:
 202          return UUID(value)
 203  
 204  
 205  class TagDateTime(JSONTag):
 206      __slots__ = ()
 207      key = " d"
 208  
 209      def check(self, value: t.Any) -> bool:
 210          return isinstance(value, datetime)
 211  
 212      def to_json(self, value: t.Any) -> t.Any:
 213          return http_date(value)
 214  
 215      def to_python(self, value: t.Any) -> t.Any:
 216          return parse_date(value)
 217  
 218  
 219  class TaggedJSONSerializer:
 220      """Serializer that uses a tag system to compactly represent objects that
 221      are not JSON types. Passed as the intermediate serializer to
 222      :class:`itsdangerous.Serializer`.
 223  
 224      The following extra types are supported:
 225  
 226      * :class:`dict`
 227      * :class:`tuple`
 228      * :class:`bytes`
 229      * :class:`~markupsafe.Markup`
 230      * :class:`~uuid.UUID`
 231      * :class:`~datetime.datetime`
 232      """
 233  
 234      __slots__ = ("tags", "order")
 235  
 236      #: Tag classes to bind when creating the serializer. Other tags can be
 237      #: added later using :meth:`~register`.
 238      default_tags = [
 239          TagDict,
 240          PassDict,
 241          TagTuple,
 242          PassList,
 243          TagBytes,
 244          TagMarkup,
 245          TagUUID,
 246          TagDateTime,
 247      ]
 248  
 249      def __init__(self) -> None:
 250          self.tags: dict[str, JSONTag] = {}
 251          self.order: list[JSONTag] = []
 252  
 253          for cls in self.default_tags:
 254              self.register(cls)
 255  
 256      def register(
 257          self,
 258          tag_class: type[JSONTag],
 259          force: bool = False,
 260          index: int | None = None,
 261      ) -> None:
 262          """Register a new tag with this serializer.
 263  
 264          :param tag_class: tag class to register. Will be instantiated with this
 265              serializer instance.
 266          :param force: overwrite an existing tag. If false (default), a
 267              :exc:`KeyError` is raised.
 268          :param index: index to insert the new tag in the tag order. Useful when
 269              the new tag is a special case of an existing tag. If ``None``
 270              (default), the tag is appended to the end of the order.
 271  
 272          :raise KeyError: if the tag key is already registered and ``force`` is
 273              not true.
 274          """
 275          tag = tag_class(self)
 276          key = tag.key
 277  
 278          if key:
 279              if not force and key in self.tags:
 280                  raise KeyError(f"Tag '{key}' is already registered.")
 281  
 282              self.tags[key] = tag
 283  
 284          if index is None:
 285              self.order.append(tag)
 286          else:
 287              self.order.insert(index, tag)
 288  
 289      def tag(self, value: t.Any) -> t.Any:
 290          """Convert a value to a tagged representation if necessary."""
 291          for tag in self.order:
 292              if tag.check(value):
 293                  return tag.tag(value)
 294  
 295          return value
 296  
 297      def untag(self, value: dict[str, t.Any]) -> t.Any:
 298          """Convert a tagged representation back to the original type."""
 299          if len(value) != 1:
 300              return value
 301  
 302          key = next(iter(value))
 303  
 304          if key not in self.tags:
 305              return value
 306  
 307          return self.tags[key].to_python(value[key])
 308  
 309      def _untag_scan(self, value: t.Any) -> t.Any:
 310          if isinstance(value, dict):
 311              # untag each item recursively
 312              value = {k: self._untag_scan(v) for k, v in value.items()}
 313              # untag the dict itself
 314              value = self.untag(value)
 315          elif isinstance(value, list):
 316              # untag each item recursively
 317              value = [self._untag_scan(item) for item in value]
 318  
 319          return value
 320  
 321      def dumps(self, value: t.Any) -> str:
 322          """Tag the value and dump it to a compact JSON string."""
 323          return dumps(self.tag(value), separators=(",", ":"))
 324  
 325      def loads(self, value: str) -> t.Any:
 326          """Load data from a JSON string and deserialized any tagged objects."""
 327          return self._untag_scan(loads(value))
 328