Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/serializers.py
Show First 20 Lines • Show All 51 Lines • ▼ Show 20 Lines | |||||
def decode_response(response: Response, extra_decoders=None) -> Any: | def decode_response(response: Response, extra_decoders=None) -> Any: | ||||
content_type = response.headers['content-type'] | content_type = response.headers['content-type'] | ||||
if content_type.startswith('application/x-msgpack'): | if content_type.startswith('application/x-msgpack'): | ||||
r = msgpack_loads(response.content, | r = msgpack_loads(response.content, | ||||
extra_decoders=extra_decoders) | extra_decoders=extra_decoders) | ||||
elif content_type.startswith('application/json'): | elif content_type.startswith('application/json'): | ||||
r = json.loads(response.text, cls=SWHJSONDecoder, | r = json_loads(response.text, | ||||
extra_decoders=extra_decoders) | extra_decoders=extra_decoders) | ||||
elif content_type.startswith('text/'): | elif content_type.startswith('text/'): | ||||
r = response.text | r = response.text | ||||
else: | else: | ||||
raise ValueError('Wrong content type `%s` for API response' | raise ValueError('Wrong content type `%s` for API response' | ||||
% content_type) | % content_type) | ||||
return r | return r | ||||
▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Lines | def decode_data(self, o: Any) -> Any: | ||||
else: | else: | ||||
return o | return o | ||||
def raw_decode(self, s: str, idx: int = 0) -> Tuple[Any, int]: | def raw_decode(self, s: str, idx: int = 0) -> Tuple[Any, int]: | ||||
data, index = super().raw_decode(s, idx) | data, index = super().raw_decode(s, idx) | ||||
return self.decode_data(data), index | return self.decode_data(data), index | ||||
def json_dumps(data: Any, extra_encoders=None) -> str: | |||||
return json.dumps(data, cls=SWHJSONEncoder, | |||||
extra_encoders=extra_encoders) | |||||
def json_loads(data: str, extra_decoders=None) -> Any: | |||||
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 = ENCODERS | ||||
if extra_encoders: | if extra_encoders: | ||||
encoders += extra_encoders | encoders += extra_encoders | ||||
def encode_types(obj): | def encode_types(obj): | ||||
if isinstance(obj, types.GeneratorType): | if isinstance(obj, types.GeneratorType): | ||||
Show All 32 Lines |