Changeset View
Changeset View
Standalone View
Standalone View
swh/web/common/converters.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 Affero General Public License version 3, or any later version | # License: GNU Affero 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 datetime | import datetime | ||||
| import json | import json | ||||
| from typing import Any, Dict, Union | from typing import Any, Dict, Union | ||||
| from swh.core.utils import decode_with_escape | from swh.core.utils import decode_with_escape | ||||
| from swh.model import hashutil | from swh.model import hashutil | ||||
| from swh.model.model import Release, Revision | from swh.model.model import RawExtrinsicMetadata, Release, Revision | ||||
| from swh.storage.interface import PartialBranches | from swh.storage.interface import PartialBranches | ||||
| from swh.web.common.typing import OriginInfo, OriginVisitInfo | from swh.web.common.typing import OriginInfo, OriginVisitInfo | ||||
| def _group_checksums(data): | def _group_checksums(data): | ||||
| """Groups checksums values computed from hash functions used in swh | """Groups checksums values computed from hash functions used in swh | ||||
| and stored in data dict under a single entry 'checksums' | and stored in data dict under a single entry 'checksums' | ||||
| """ | """ | ||||
| ▲ Show 20 Lines • Show All 220 Lines • ▼ Show 20 Lines | def default(self, obj): | ||||
| return obj.decode("utf-8") | return obj.decode("utf-8") | ||||
| except UnicodeDecodeError: | except UnicodeDecodeError: | ||||
| # fallback to binary representation to avoid display errors | # fallback to binary representation to avoid display errors | ||||
| return repr(obj) | return repr(obj) | ||||
| # Let the base class default method raise the TypeError | # Let the base class default method raise the TypeError | ||||
| return json.JSONEncoder.default(self, obj) | return json.JSONEncoder.default(self, obj) | ||||
| def convert_revision_metadata(metadata): | def convert_metadata(metadata): | ||||
| """Convert json specific dict to a json serializable one. | """Convert json specific dict to a json serializable one. | ||||
| """ | """ | ||||
| if not metadata: | if metadata is None: | ||||
| return {} | return {} | ||||
| return json.loads(json.dumps(metadata, cls=SWHMetadataEncoder)) | return json.loads(json.dumps(metadata, cls=SWHMetadataEncoder)) | ||||
| def from_revision(revision: Union[Dict[str, Any], Revision]) -> Dict[str, Any]: | def from_revision(revision: Union[Dict[str, Any], Revision]) -> Dict[str, Any]: | ||||
| """Convert swh revision model object to a json serializable revision dictionary. | """Convert swh revision model object to a json serializable revision dictionary. | ||||
| Show All 14 Lines | if isinstance(revision, Revision): | ||||
| revision_d = revision.to_dict() | revision_d = revision.to_dict() | ||||
| else: | else: | ||||
| revision_d = revision | revision_d = revision | ||||
| revision_d = from_swh( | revision_d = from_swh( | ||||
| revision_d, | revision_d, | ||||
| hashess={"id", "directory", "parents", "children"}, | hashess={"id", "directory", "parents", "children"}, | ||||
| bytess={"name", "fullname", "email", "extra_headers", "message"}, | bytess={"name", "fullname", "email", "extra_headers", "message"}, | ||||
| convert={"metadata"}, | convert={"metadata"}, | ||||
| convert_fn=convert_revision_metadata, | convert_fn=convert_metadata, | ||||
| dates={"date", "committer_date"}, | dates={"date", "committer_date"}, | ||||
| ) | ) | ||||
| if revision_d: | if revision_d: | ||||
| if "parents" in revision_d: | if "parents" in revision_d: | ||||
| revision_d["merge"] = len(revision_d["parents"]) > 1 | revision_d["merge"] = len(revision_d["parents"]) > 1 | ||||
| return revision_d | return revision_d | ||||
| def from_raw_extrinsic_metadata( | |||||
| metadata: Union[Dict[str, Any], RawExtrinsicMetadata] | |||||
| ) -> Dict[str, Any]: | |||||
| """Convert RawExtrinsicMetadata model object to a json serializable dictionary. | |||||
| """ | |||||
| return from_swh( | |||||
| metadata.to_dict() if isinstance(metadata, RawExtrinsicMetadata) else metadata, | |||||
| blacklist={"id", "metadata"}, | |||||
| dates={"discovery_date"}, | |||||
| ) | |||||
| def from_content(content): | def from_content(content): | ||||
ardumont: there, compacted ;) | |||||
| """Convert swh content to serializable content dictionary. | """Convert swh content to serializable content dictionary. | ||||
| """ | """ | ||||
| return from_swh( | return from_swh( | ||||
| content, | content, | ||||
| hashess={"sha1", "sha1_git", "sha256", "blake2s256"}, | hashess={"sha1", "sha1_git", "sha256", "blake2s256"}, | ||||
| blacklist={"ctime"}, | blacklist={"ctime"}, | ||||
| convert={"status"}, | convert={"status"}, | ||||
| ▲ Show 20 Lines • Show All 80 Lines • Show Last 20 Lines | |||||
there, compacted ;)