swh.core.api.serializers module

swh.core.api.serializers.encode_datetime(dt: datetime.datetime) → str[source]

Wrapper of datetime.datetime.isoformat() that forbids naive datetimes.

class swh.core.api.serializers.MsgpackExtTypeCodes(value)[source]

Bases: enum.Enum

An enumeration.

LONG_INT = 1
swh.core.api.serializers.encode_data_client(data: Any, extra_encoders=None) → bytes[source]
swh.core.api.serializers.decode_response(response: requests.models.Response, extra_decoders=None) → Any[source]
class swh.core.api.serializers.SWHJSONEncoder(extra_encoders=None, **kwargs)[source]

Bases: json.encoder.JSONEncoder

JSON encoder for data structures generated by Software Heritage.

This JSON encoder extends the default Python JSON encoder and adds awareness for the following specific types:

  • bytes (get encoded as a Base85 string);

  • datetime.datetime (get encoded as an ISO8601 string).

Non-standard types get encoded as a a dictionary with two keys:

  • swhtype with value ‘bytes’ or ‘datetime’;

  • d containing the encoded value.

SWHJSONEncoder also encodes arbitrary iterables as a list (allowing serialization of generators).

Caveats: Limitations in the JSONEncoder extension mechanism prevent us from “escaping” dictionaries that only contain the swhtype and d keys, and therefore arbitrary data structures can’t be round-tripped through SWHJSONEncoder and SWHJSONDecoder.

default(o: Any) → Union[Dict[str, Union[Dict[str, int], str]], list][source]

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)
class swh.core.api.serializers.SWHJSONDecoder(extra_decoders=None, **kwargs)[source]

Bases: json.decoder.JSONDecoder

JSON decoder for data structures encoded with SWHJSONEncoder.

This JSON decoder extends the default Python JSON decoder, allowing the decoding of:

  • bytes (encoded as a Base85 string);

  • datetime.datetime (encoded as an ISO8601 string).

Non-standard types must be encoded as a a dictionary with exactly two keys:

  • swhtype with value ‘bytes’ or ‘datetime’;

  • d containing the encoded value.

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

decode_data(o: Any) → Any[source]
raw_decode(s: str, idx: int = 0) → Tuple[Any, int][source]

Decode a JSON document from s (a str beginning with a JSON document) and return a 2-tuple of the Python representation and the index in s where the document ended.

This can be used to decode a JSON document from a string that may have extraneous data at the end.

swh.core.api.serializers.json_dumps(data: Any, extra_encoders=None) → str[source]
swh.core.api.serializers.json_loads(data: str, extra_decoders=None) → Any[source]
swh.core.api.serializers.msgpack_dumps(data: Any, extra_encoders=None) → bytes[source]

Write data as a msgpack stream

swh.core.api.serializers.msgpack_loads(data: bytes, extra_decoders=None) → Any[source]

Read data as a msgpack stream.

Caution

This function is used by swh.journal to decode the contents of the journal. This function must be kept backwards-compatible.

swh.core.api.serializers.exception_to_dict(exception)[source]