Differential D4684 Diff 16630 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, Callable, Dict, List, Optional, Tuple, Type, Union		from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
from uuid import UUID		from uuid import UUID

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


def encode_datetime(dt: datetime.datetime) -> str:		def encode_datetime(dt: datetime.datetime) -> str:
Show All 28 Lines	def exception_to_dict(exception: Exception) -> Dict[str, Any]:
}		}


def dict_to_exception(exc_dict: Dict[str, Any]) -> Exception:		def dict_to_exception(exc_dict: Dict[str, Any]) -> Exception:
temp = __import__(exc_dict["module"], fromlist=[exc_dict["type"]])		temp = __import__(exc_dict["module"], fromlist=[exc_dict["type"]])
return getattr(temp, exc_dict["type"])(*exc_dict["args"])		return getattr(temp, exc_dict["type"])(*exc_dict["args"])


ENCODERS = [		def encode_timedelta(td: datetime.timedelta) -> Dict[str, int]:
(arrow.Arrow, "arrow", arrow.Arrow.isoformat),		return {
		"days": td.days,
		"seconds": td.seconds,
		"microseconds": td.microseconds,
		}


		ENCODERS: List[Tuple[type, str, Callable]] = [
(datetime.datetime, "datetime", encode_datetime),		(datetime.datetime, "datetime", encode_datetime),
(
datetime.timedelta,
"timedelta",
lambda o: {
"days": o.days,
"seconds": o.seconds,
"microseconds": o.microseconds,
},
),
(UUID, "uuid", str),		(UUID, "uuid", str),
		(datetime.timedelta, "timedelta", encode_timedelta),
(PagedResult, "paged_result", _encode_paged_result),		(PagedResult, "paged_result", _encode_paged_result),
# Only for JSON:		# Only for JSON:
(bytes, "bytes", lambda o: base64.b85encode(o).decode("ascii")),		(bytes, "bytes", lambda o: base64.b85encode(o).decode("ascii")),
(Exception, "exception", exception_to_dict),		(Exception, "exception", exception_to_dict),
]		]

DECODERS = {		DECODERS: Dict[str, Callable] = {
		olasdUnsubmitted Done Inline Actions I think `Dict[str, Callable]` should be enough, actually. olasd: I think `Dict[str, Callable]` should be enough, actually.
		douarddaAuthorUnsubmitted Done Inline Actions ok let's try it (thx) douardda: ok let's try it (thx)
"arrow": arrow.get,
"datetime": lambda d: iso8601.parse_date(d, default_timezone=None),		"datetime": lambda d: iso8601.parse_date(d, default_timezone=None),
"timedelta": lambda d: datetime.timedelta(**d),		"timedelta": lambda d: datetime.timedelta(**d),
"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,
}		}
▲ Show 20 Lines • Show All 216 Lines • Show Last 20 Lines