Changeset View
Changeset View
Standalone View
Standalone View
swh/journal/serializers.py
# Copyright (C) 2016-2017 The Software Heritage developers | # Copyright (C) 2016-2017 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 | ||||
from typing import Any, Dict, Union, overload | from typing import Any, Dict, Union, overload | ||||
import msgpack | import msgpack | ||||
from swh.core.api.serializers import msgpack_dumps, msgpack_loads | from swh.core.api.serializers import msgpack_dumps, msgpack_loads | ||||
from swh.model.hashutil import DEFAULT_ALGORITHMS | from swh.model.hashutil import DEFAULT_ALGORITHMS | ||||
from swh.model.model import ( | from swh.model.model import ( | ||||
Content, | Content, | ||||
Directory, | Directory, | ||||
Origin, | Origin, | ||||
OriginVisit, | OriginVisit, | ||||
OriginVisitStatus, | |||||
Release, | Release, | ||||
Revision, | Revision, | ||||
SkippedContent, | SkippedContent, | ||||
Snapshot, | Snapshot, | ||||
) | ) | ||||
ModelObject = Union[ | ModelObject = Union[ | ||||
Content, Directory, Origin, OriginVisit, Release, Revision, SkippedContent, Snapshot | Content, | ||||
Directory, | |||||
Origin, | |||||
OriginVisit, | |||||
OriginVisitStatus, | |||||
Release, | |||||
Revision, | |||||
SkippedContent, | |||||
Snapshot, | |||||
] | ] | ||||
KeyType = Union[Dict[str, str], Dict[str, bytes], bytes] | KeyType = Union[Dict[str, str], Dict[str, bytes], bytes] | ||||
# these @overload'ed versions of the object_key method aim at helping mypy figuring | # these @overload'ed versions of the object_key method aim at helping mypy figuring | ||||
# the correct type-ing. | # the correct type-ing. | ||||
@overload | @overload | ||||
def object_key( | def object_key( | ||||
object_type: str, object_: Union[Content, Directory, Revision, Release, Snapshot] | object_type: str, object_: Union[Content, Directory, Revision, Release, Snapshot] | ||||
) -> bytes: | ) -> bytes: | ||||
... | ... | ||||
@overload | @overload | ||||
def object_key( | def object_key( | ||||
object_type: str, object_: Union[Origin, SkippedContent] | object_type: str, object_: Union[Origin, SkippedContent] | ||||
) -> Dict[str, bytes]: | ) -> Dict[str, bytes]: | ||||
... | ... | ||||
@overload | @overload | ||||
def object_key(object_type: str, object_: OriginVisit) -> Dict[str, str]: | def object_key( | ||||
object_type: str, object_: Union[OriginVisit, OriginVisitStatus] | |||||
) -> Dict[str, str]: | |||||
... | ... | ||||
def object_key(object_type: str, object_) -> KeyType: | def object_key(object_type: str, object_) -> KeyType: | ||||
if object_type in ("revision", "release", "directory", "snapshot"): | if object_type in ("revision", "release", "directory", "snapshot"): | ||||
return object_.id | return object_.id | ||||
elif object_type == "content": | elif object_type == "content": | ||||
return object_.sha1 # TODO: use a dict of hashes | return object_.sha1 # TODO: use a dict of hashes | ||||
▲ Show 20 Lines • Show All 72 Lines • Show Last 20 Lines |