tag.py 8.6 KB

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