Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/utils.py
# Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-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 | ||||
from types import GeneratorType | from types import GeneratorType | ||||
from typing import Any, Dict, Tuple, Union | |||||
import iso8601 | import iso8601 | ||||
from swh.model.identifiers import normalize_timestamp | from swh.model.identifiers import SWHID, normalize_timestamp, parse_swhid | ||||
from swh.model.model import MetadataTargetType | |||||
def merge(*dicts): | def merge(*dicts): | ||||
"""Given an iterator of dicts, merge them losing no information. | """Given an iterator of dicts, merge them losing no information. | ||||
Args: | Args: | ||||
*dicts: arguments are all supposed to be dict to merge into one | *dicts: arguments are all supposed to be dict to merge into one | ||||
▲ Show 20 Lines • Show All 57 Lines • ▼ Show 20 Lines | def normalize_date(date): | ||||
""" | """ | ||||
if isinstance(date, list): | if isinstance(date, list): | ||||
date = date[0] | date = date[0] | ||||
if isinstance(date, str): | if isinstance(date, str): | ||||
date = iso8601.parse_date(date) | date = iso8601.parse_date(date) | ||||
return normalize_timestamp(date) | return normalize_timestamp(date) | ||||
def compute_metadata_context( | |||||
swhid_reference: Union[SWHID, str] | |||||
) -> Tuple[MetadataTargetType, Dict[str, Any]]: | |||||
"""Given a SWHID object, determine the context as a dict. | |||||
The parse_swhid calls within are not expected to raise (because they should have | |||||
been caught early on). | |||||
""" | |||||
metadata_context: Dict[str, Any] = {"origin": None} | |||||
if isinstance(swhid_reference, SWHID): | |||||
object_type = MetadataTargetType(swhid_reference.object_type) | |||||
assert object_type != MetadataTargetType.ORIGIN | |||||
if swhid_reference.metadata: | |||||
path = swhid_reference.metadata.get("path") | |||||
metadata_context = { | |||||
"origin": swhid_reference.metadata.get("origin"), | |||||
"path": path.encode() if path else None, | |||||
} | |||||
snapshot = swhid_reference.metadata.get("visit") | |||||
if snapshot: | |||||
moranegg: isn't it called `visit` in the SWHID context? is that different on the ERMDS? | |||||
Not Done Inline Actionsyes and yes. vlorentz: yes and yes. | |||||
metadata_context["snapshot"] = parse_swhid(snapshot) | |||||
anchor = swhid_reference.metadata.get("anchor") | |||||
if anchor: | |||||
anchor_swhid = parse_swhid(anchor) | |||||
Not Done Inline Actionsthis is called anchor in a SWHID. moranegg: this is called `anchor` in a SWHID.
Looks like the context names sent to the ERMDS is a bit… | |||||
Not Done Inline Actions
yes and it's intentional
what do you mean? vlorentz: > this is called anchor in a SWHID.
> Looks like the context names sent to the ERMDS is a bit… | |||||
metadata_context[anchor_swhid.object_type] = anchor_swhid | |||||
else: | |||||
object_type = MetadataTargetType.ORIGIN | |||||
return object_type, metadata_context |
isn't it called visit in the SWHID context? is that different on the ERMDS?