Differential D4572 Diff 16253 swh/core/api/serializers.py

Changeset View

Standalone View

swh/core/api/serializers.py

# Copyright (C) 2015-2020 The Software Heritage developers		# Copyright (C) 2015-2020 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version		# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

import base64		import base64
import datetime		import datetime
from enum import Enum		from enum import Enum
import json		import json
import traceback		import traceback
import types		import types
from typing import Any, Dict, Tuple, Union		from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
from uuid import UUID		from uuid import UUID

import arrow		import arrow
import iso8601		import iso8601
import msgpack		import msgpack
from requests import Response		from requests import Response

from swh.core.api.classes import PagedResult		from swh.core.api.classes import PagedResult
▲ Show 20 Lines • Show All 62 Lines • ▼ Show 20 Lines	DECODERS = {
"uuid": UUID,		"uuid": UUID,
"paged_result": _decode_paged_result,		"paged_result": _decode_paged_result,
# Only for JSON:		# Only for JSON:
"bytes": base64.b85decode,		"bytes": base64.b85decode,
"exception": dict_to_exception,		"exception": dict_to_exception,
}		}


		def get_encoders(
		extra_encoders: Optional[List[Tuple[Type, str, Callable]]]
		) -> List[Tuple[Type, str, Callable]]:
		if extra_encoders is not None:
		return [ENCODERS, extra_encoders]
		else:
		return ENCODERS


		def get_decoders(extra_decoders: Optional[Dict[str, Callable]]) -> Dict[str, Callable]:
		if extra_decoders is not None:
		return {DECODERS, extra_decoders}
		else:
		return DECODERS


class MsgpackExtTypeCodes(Enum):		class MsgpackExtTypeCodes(Enum):
LONG_INT = 1		LONG_INT = 1
LONG_NEG_INT = 2		LONG_NEG_INT = 2


def encode_data_client(data: Any, extra_encoders=None) -> bytes:		def encode_data_client(data: Any, extra_encoders=None) -> bytes:
try:		try:
return msgpack_dumps(data, extra_encoders=extra_encoders)		return msgpack_dumps(data, extra_encoders=extra_encoders)
Show All 37 Lines	class SWHJSONEncoder(json.JSONEncoder):
prevent us from "escaping" dictionaries that only contain the		prevent us from "escaping" dictionaries that only contain the
swhtype and d keys, and therefore arbitrary data structures can't		swhtype and d keys, and therefore arbitrary data structures can't
be round-tripped through SWHJSONEncoder and SWHJSONDecoder.		be round-tripped through SWHJSONEncoder and SWHJSONDecoder.

"""		"""

def __init__(self, extra_encoders=None, **kwargs):		def __init__(self, extra_encoders=None, **kwargs):
super().__init__(**kwargs)		super().__init__(**kwargs)
self.encoders = ENCODERS		self.encoders = get_encoders(extra_encoders)
if extra_encoders:
self.encoders += extra_encoders

def default(self, o: Any) -> Union[Dict[str, Union[Dict[str, int], str]], list]:		def default(self, o: Any) -> Union[Dict[str, Union[Dict[str, int], str]], list]:
for (type_, type_name, encoder) in self.encoders:		for (type_, type_name, encoder) in self.encoders:
if isinstance(o, type_):		if isinstance(o, type_):
return {		return {
"swhtype": type_name,		"swhtype": type_name,
"d": encoder(o),		"d": encoder(o),
}		}
Show All 25 Lines	class SWHJSONDecoder(json.JSONDecoder):

To limit the impact our encoding, if the swhtype key doesn't		To limit the impact our encoding, if the swhtype key doesn't
contain a known value, the dictionary is decoded as-is.		contain a known value, the dictionary is decoded as-is.

"""		"""

def __init__(self, extra_decoders=None, **kwargs):		def __init__(self, extra_decoders=None, **kwargs):
super().__init__(**kwargs)		super().__init__(**kwargs)
self.decoders = DECODERS		self.decoders = get_decoders(extra_decoders)
if extra_decoders:
self.decoders = {self.decoders, extra_decoders}

def decode_data(self, o: Any) -> Any:		def decode_data(self, o: Any) -> Any:
if isinstance(o, dict):		if isinstance(o, dict):
if set(o.keys()) == {"d", "swhtype"}:		if set(o.keys()) == {"d", "swhtype"}:
if o["swhtype"] == "bytes":		if o["swhtype"] == "bytes":
return base64.b85decode(o["d"])		return base64.b85decode(o["d"])
decoder = self.decoders.get(o["swhtype"])		decoder = self.decoders.get(o["swhtype"])
if decoder:		if decoder:
Show All 14 Lines


def json_loads(data: str, extra_decoders=None) -> Any:		def json_loads(data: str, extra_decoders=None) -> Any:
return json.loads(data, cls=SWHJSONDecoder, extra_decoders=extra_decoders)		return json.loads(data, cls=SWHJSONDecoder, extra_decoders=extra_decoders)


def msgpack_dumps(data: Any, extra_encoders=None) -> bytes:		def msgpack_dumps(data: Any, extra_encoders=None) -> bytes:
"""Write data as a msgpack stream"""		"""Write data as a msgpack stream"""
encoders = ENCODERS		encoders = get_encoders(extra_encoders)
if extra_encoders:
encoders += extra_encoders

def encode_types(obj):		def encode_types(obj):
if isinstance(obj, int):		if isinstance(obj, int):
# integer overflowed while packing. Handle it as an extended type		# integer overflowed while packing. Handle it as an extended type
if obj > 0:		if obj > 0:
code = MsgpackExtTypeCodes.LONG_INT.value		code = MsgpackExtTypeCodes.LONG_INT.value
else:		else:
code = MsgpackExtTypeCodes.LONG_NEG_INT.value		code = MsgpackExtTypeCodes.LONG_NEG_INT.value
Show All 19 Lines

def msgpack_loads(data: bytes, extra_decoders=None) -> Any:		def msgpack_loads(data: bytes, extra_decoders=None) -> Any:
"""Read data as a msgpack stream.		"""Read data as a msgpack stream.

.. Caution::		.. Caution::
This function is used by swh.journal to decode the contents of the		This function is used by swh.journal to decode the contents of the
journal. This function must be kept backwards-compatible.		journal. This function must be kept backwards-compatible.
"""		"""
decoders = DECODERS		decoders = get_decoders(extra_decoders)
if extra_decoders:
decoders = {decoders, extra_decoders}

def ext_hook(code, data):		def ext_hook(code, data):
if code == MsgpackExtTypeCodes.LONG_INT.value:		if code == MsgpackExtTypeCodes.LONG_INT.value:
return int.from_bytes(data, "big")		return int.from_bytes(data, "big")
elif code == MsgpackExtTypeCodes.LONG_NEG_INT.value:		elif code == MsgpackExtTypeCodes.LONG_NEG_INT.value:
return -int.from_bytes(data, "big")		return -int.from_bytes(data, "big")
raise ValueError("Unknown msgpack extended code %s" % code)		raise ValueError("Unknown msgpack extended code %s" % code)

Show All 37 Lines