Changeset View
Changeset View
Standalone View
Standalone View
swh/provenance/storage/interface.py
# Copyright (C) 2021-2022 The Software Heritage developers | # Copyright (C) 2021-2022 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 __future__ import annotations | from __future__ import annotations | ||||
from dataclasses import dataclass | from dataclasses import dataclass | ||||
from datetime import datetime | from datetime import datetime | ||||
import enum | import enum | ||||
from types import TracebackType | from types import TracebackType | ||||
from typing import Dict, Generator, Iterable, List, Optional, Set, Type, Union | from typing import Dict, Generator, Iterable, List, Optional, Set, Type | ||||
from typing_extensions import Protocol, runtime_checkable | from typing_extensions import Protocol, runtime_checkable | ||||
from swh.core.api import remote_api_endpoint | from swh.core.api import remote_api_endpoint | ||||
from swh.model.model import Sha1Git | from swh.model.model import Sha1Git | ||||
class EntityType(enum.Enum): | class EntityType(enum.Enum): | ||||
▲ Show 20 Lines • Show All 125 Lines • ▼ Show 20 Lines | class ProvenanceStorageInterface(Protocol): | ||||
@remote_api_endpoint("entity_get_all") | @remote_api_endpoint("entity_get_all") | ||||
def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]: | def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]: | ||||
"""Retrieve all sha1 ids for entities of type `entity` present in the provenance | """Retrieve all sha1 ids for entities of type `entity` present in the provenance | ||||
model. This method is used only in tests. | model. This method is used only in tests. | ||||
""" | """ | ||||
... | ... | ||||
@remote_api_endpoint("location_add") | @remote_api_endpoint("location_add") | ||||
def location_add(self, paths: Iterable[bytes]) -> bool: | def location_add(self, paths: Dict[Sha1Git, bytes]) -> bool: | ||||
"""Register the given `paths` in the storage.""" | """Register the given `paths` in the storage.""" | ||||
... | ... | ||||
@remote_api_endpoint("location_get_all") | @remote_api_endpoint("location_get_all") | ||||
def location_get_all(self) -> Set[bytes]: | def location_get_all(self) -> Dict[Sha1Git, bytes]: | ||||
"""Retrieve all paths present in the provenance model. | """Retrieve all paths present in the provenance model. | ||||
This method is used only in tests.""" | This method is used only in tests.""" | ||||
... | ... | ||||
@remote_api_endpoint("open") | @remote_api_endpoint("open") | ||||
def open(self) -> None: | def open(self) -> None: | ||||
"""Open connection to the storage and allocate necessary resources.""" | """Open connection to the storage and allocate necessary resources.""" | ||||
... | ... | ||||
@remote_api_endpoint("origin_add") | @remote_api_endpoint("origin_add") | ||||
def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool: | def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool: | ||||
"""Add origins identified by sha1 ids, with their corresponding url (as paired | """Add origins identified by sha1 ids, with their corresponding url (as paired | ||||
in `orgs`) to the provenance storage. Return a boolean stating if the | in `orgs`) to the provenance storage. Return a boolean stating if the | ||||
information was successfully stored. | information was successfully stored. | ||||
""" | """ | ||||
... | ... | ||||
@remote_api_endpoint("origin_get") | @remote_api_endpoint("origin_get") | ||||
def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]: | def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]: | ||||
"""Retrieve the associated url for each origin sha1 in `ids`.""" | """Retrieve the associated url for each origin sha1 in `ids`.""" | ||||
... | ... | ||||
@remote_api_endpoint("revision_add") | @remote_api_endpoint("revision_add") | ||||
def revision_add( | def revision_add(self, revs: Dict[Sha1Git, RevisionData]) -> bool: | ||||
self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]] | |||||
) -> bool: | |||||
"""Add revisions identified by sha1 ids, with optional associated date or origin | """Add revisions identified by sha1 ids, with optional associated date or origin | ||||
(as paired in `revs`) to the provenance storage. Return a boolean stating if the | (as paired in `revs`) to the provenance storage. Return a boolean stating if the | ||||
information was successfully stored. | information was successfully stored. | ||||
""" | """ | ||||
... | ... | ||||
@remote_api_endpoint("revision_get") | @remote_api_endpoint("revision_get") | ||||
def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]: | def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]: | ||||
Show All 38 Lines |