Browse Source

pass serializer at tag init instead of to each method

split tagged dict and passthrough into separate cases
add docstrings
pull/2352/head
David Lord 8 years ago
parent
commit
ca176cb903
No known key found for this signature in database
GPG Key ID: 7A1C87E3F5BC42A8
  1. 201
      flask/json/tag.py

201
flask/json/tag.py

@ -10,179 +10,244 @@ from flask.json import dumps, loads
class JSONTag(object): class JSONTag(object):
__slots__ = () """Base class for defining type tags for :class:`TaggedJSONSerializer`."""
__slots__ = ('serializer',)
key = None key = None
"""The tag to mark the serialized object with. If ``None``, this tag is
only used as an intermediate step during tagging."""
def __init__(self, serializer):
"""Create a tagger for the given serializer."""
self.serializer = serializer
def check(self, value):
"""Check if the given value should be tagged by this tag."""
def check(self, serializer, value):
raise NotImplementedError raise NotImplementedError
def to_json(self, serializer, value): def to_json(self, value):
"""Convert the Python object to an object that is a valid JSON type.
The tag will be added later."""
raise NotImplementedError raise NotImplementedError
def to_python(self, serializer, value): def to_python(self, value):
"""Convert the JSON representation back to the correct type. The tag
will already be removed."""
raise NotImplementedError raise NotImplementedError
def tag(self, serializer, value): def tag(self, value):
return {self.key: self.to_json(serializer, value)} """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): class TagDict(JSONTag):
__slots__ = () """Tag for 1-item dicts whose only key matches a registered tag.
key = ' di'
def check(self, serializer, value): Internally, the dict key is suffixed with `__`, and the suffix is removed
return isinstance(value, dict) when deserializing.
"""
def to_json(self, serializer, value, key=None): __slots__ = ('serializer',)
if key is not None: key = ' di'
return {key + '__': serializer._tag(value[key])}
return dict((k, serializer._tag(v)) for k, v in iteritems(value)) def check(self, value):
return (
isinstance(value, dict)
and len(value) == 1
and next(iter(value)) in self.serializer.tags
)
def to_python(self, serializer, value): def to_json(self, value):
key = next(iter(value)) key = next(iter(value))
return {key[:-2]: value[key]} return {key + '__': self.serializer.tag(value[key])}
def tag(self, serializer, value): def to_python(self, value):
if len(value) == 1:
key = next(iter(value)) key = next(iter(value))
return {key[:-2]: value[key]}
if key in serializer._tags:
return {self.key: self.to_json(serializer, value, key=key)}
return self.to_json(serializer, value) class PassDict(JSONTag):
__slots__ = ('serializer',)
def check(self, value):
return isinstance(value, dict)
def to_json(self, value):
# JSON objects may only have string keys, so don't bother tagging the
# key here.
return dict((k, self.serializer.tag(v)) for k, v in iteritems(value))
tag = to_json
class TagTuple(JSONTag): class TagTuple(JSONTag):
__slots__ = () __slots__ = ('serializer',)
key = ' t' key = ' t'
def check(self, serializer, value): def check(self, value):
return isinstance(value, tuple) return isinstance(value, tuple)
def to_json(self, serializer, value): def to_json(self, value):
return [serializer._tag(item) for item in value] return [self.serializer.tag(item) for item in value]
def to_python(self, serializer, value): def to_python(self, value):
return tuple(value) return tuple(value)
class PassList(JSONTag): class PassList(JSONTag):
__slots__ = () __slots__ = ('serializer',)
def check(self, serializer, value): def check(self, value):
return isinstance(value, list) return isinstance(value, list)
def to_json(self, serializer, value): def to_json(self, value):
return [serializer._tag(item) for item in value] return [self.serializer.tag(item) for item in value]
tag = to_json tag = to_json
class TagBytes(JSONTag): class TagBytes(JSONTag):
__slots__ = () __slots__ = ('serializer',)
key = ' b' key = ' b'
def check(self, serializer, value): def check(self, value):
return isinstance(value, bytes) return isinstance(value, bytes)
def to_json(self, serializer, value): def to_json(self, value):
return b64encode(value).decode('ascii') return b64encode(value).decode('ascii')
def to_python(self, serializer, value): def to_python(self, value):
return b64decode(value) return b64decode(value)
class TagMarkup(JSONTag): class TagMarkup(JSONTag):
__slots__ = () """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__ = ('serializer',)
key = ' m' key = ' m'
def check(self, serializer, value): def check(self, value):
return callable(getattr(value, '__html__', None)) return callable(getattr(value, '__html__', None))
def to_json(self, serializer, value): def to_json(self, value):
return text_type(value.__html__()) return text_type(value.__html__())
def to_python(self, serializer, value): def to_python(self, value):
return Markup(value) return Markup(value)
class TagUUID(JSONTag): class TagUUID(JSONTag):
__slots__ = () __slots__ = ('serializer',)
key = ' u' key = ' u'
def check(self, serializer, value): def check(self, value):
return isinstance(value, UUID) return isinstance(value, UUID)
def to_json(self, serializer, value): def to_json(self, value):
return value.hex return value.hex
def to_python(self, serializer, value): def to_python(self, value):
return UUID(value) return UUID(value)
class TagDateTime(JSONTag): class TagDateTime(JSONTag):
__slots__ = () __slots__ = ('serializer',)
key = ' d' key = ' d'
def check(self, serializer, value): def check(self, value):
return isinstance(value, datetime) return isinstance(value, datetime)
def to_json(self, serializer, value): def to_json(self, value):
return http_date(value) return http_date(value)
def to_python(self, serializer, value): def to_python(self, value):
return parse_date(value) return parse_date(value)
class TaggedJSONSerializer(object): class TaggedJSONSerializer(object):
__slots__ = ('_tags', '_order') """Serializer that uses a tag system to compactly represent objects that
_default_tags = [ are not JSON types. Passed as the intermediate serializer to
TagDict(), TagTuple(), PassList(), TagBytes(), TagMarkup(), TagUUID(), :class:`itsdangerous.Serializer`."""
TagDateTime(),
__slots__ = ('tags', 'order')
default_tags = [
TagDict, PassDict, TagTuple, PassList, TagBytes, TagMarkup, TagUUID,
TagDateTime,
] ]
"""Tag classes to bind when creating the serializer. Other tags can be
added later using :meth:`~register`."""
def __init__(self): def __init__(self):
self._tags = {} self.tags = {}
self._order = [] self.order = []
for tag in self._default_tags: for cls in self.default_tags:
self.register(tag) self.register(cls)
def register(self, tag, force=False, index=-1): def register(self, tag_class, force=False, index=-1):
"""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 -1 (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 key = tag.key
if key is not None: if key is not None:
if not force and key in self._tags: if not force and key in self.tags:
raise KeyError("Tag '{0}' is already registered.".format(key)) raise KeyError("Tag '{0}' is already registered.".format(key))
self._tags[key] = tag self.tags[key] = tag
if index == -1: if index == -1:
self._order.append(tag) self.order.append(tag)
else: else:
self._order.insert(index, tag) self.order.insert(index, tag)
def _tag(self, value): def tag(self, value):
for tag in self._order: """Convert a value to a tagged representation if necessary."""
if tag.check(self, value):
return tag.tag(self, value) for tag in self.order:
if tag.check(value):
return tag.tag(value)
return value return value
def _untag(self, value): def untag(self, value):
"""Convert a tagged representation back to the original type."""
if len(value) != 1: if len(value) != 1:
return value return value
key = next(iter(value)) key = next(iter(value))
if key not in self._tags: if key not in self.tags:
return value return value
return self._tags[key].to_python(self, value[key]) return self.tags[key].to_python(value[key])
def dumps(self, value): def dumps(self, value):
return dumps(self._tag(value), separators=(',', ':')) """Tag the value and dump it to a compact JSON string."""
return dumps(self.tag(value), separators=(',', ':'))
def loads(self, value): def loads(self, value):
return loads(value, object_hook=self._untag) """Load data from a JSON string and deserialized any tagged objects."""
return loads(value, object_hook=self.untag)

Loading…
Cancel
Save