Differential D8656 Diff 31273 swh/provenance/storage/interface.py

Changeset 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 110 Lines • ▼ Show 20 Lines	class ProvenanceStorageInterface(Protocol):
@remote_api_endpoint("directory_get")		@remote_api_endpoint("directory_get")
def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, DirectoryData]:		def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, DirectoryData]:
"""Retrieve the associated date and (optional) flatten flag for each directory		"""Retrieve the associated date and (optional) flatten flag for each directory
sha1 in `ids`. If some directories has no associated date, it is not present in		sha1 in `ids`. If some directories has no associated date, it is not present in
the resulting dictionary.		the resulting dictionary.
"""		"""
...		...

@remote_api_endpoint("directory_iter_not_flattenned")		@remote_api_endpoint("directory_iter_not_flattened")
def directory_iter_not_flattenned(		def directory_iter_not_flattened(
self, limit: int, start_id: Sha1Git		self, limit: int, start_id: Sha1Git
) -> List[Sha1Git]:		) -> List[Sha1Git]:
"""Retrieve the unflattenned directories after ``start_id`` up to ``limit`` entries."""		"""Retrieve the unflattened directories after ``start_id`` up to ``limit`` entries."""
...		...

@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