Differential D5986 Diff 21586 swh/provenance/interface.py

Changeset View

Standalone View

swh/provenance/interface.py

# Copyright (C) 2021 The Software Heritage developers		# Copyright (C) 2021 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 dataclasses import dataclass		from dataclasses import dataclass
from datetime import datetime		from datetime import datetime
import enum		import enum
from typing import Dict, Generator, Iterable, Optional, Set		from typing import Dict, Generator, Iterable, Optional, Set

from typing_extensions import Protocol, runtime_checkable		from typing_extensions import Protocol, runtime_checkable

		from swh.core.api import remote_api_endpoint
from swh.model.model import Sha1Git		from swh.model.model import Sha1Git

from .model import DirectoryEntry, FileEntry, OriginEntry, RevisionEntry		from .model import DirectoryEntry, FileEntry, OriginEntry, RevisionEntry


class EntityType(enum.Enum):		class EntityType(enum.Enum):
CONTENT = "content"		CONTENT = "content"
DIRECTORY = "directory"		DIRECTORY = "directory"
Show All 39 Lines	class RelationData:

src: Sha1Git		src: Sha1Git
dst: Sha1Git		dst: Sha1Git
path: Optional[bytes]		path: Optional[bytes]


@runtime_checkable		@runtime_checkable
class ProvenanceStorageInterface(Protocol):		class ProvenanceStorageInterface(Protocol):
raise_on_commit: bool = False		@remote_api_endpoint("content_find_first")

def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:		def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
"""Retrieve the first occurrence of the blob identified by `id`."""		"""Retrieve the first occurrence of the blob identified by `id`."""
...		...

		@remote_api_endpoint("content_find_all")
def content_find_all(		def content_find_all(
self, id: Sha1Git, limit: Optional[int] = None		self, id: Sha1Git, limit: Optional[int] = None
) -> Generator[ProvenanceResult, None, None]:		) -> Generator[ProvenanceResult, None, None]:
"""Retrieve all the occurrences of the blob identified by `id`."""		"""Retrieve all the occurrences of the blob identified by `id`."""
...		...

		@remote_api_endpoint("content_set_date")
def content_set_date(self, dates: Dict[Sha1Git, datetime]) -> bool:		def content_set_date(self, dates: Dict[Sha1Git, datetime]) -> bool:
"""Associate dates to blobs identified by sha1 ids, as paired in `dates`. Return		"""Associate dates to blobs identified by sha1 ids, as paired in `dates`. Return
a boolean stating whether the information was successfully stored.		a boolean stating whether the information was successfully stored.
"""		"""
...		...

		@remote_api_endpoint("content_get")
def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
"""Retrieve the associated date for each blob sha1 in `ids`. If some blob has		"""Retrieve the associated date for each blob sha1 in `ids`. If some blob has
no associated date, it is not present in the resulting dictionary.		no associated date, it is not present in the resulting dictionary.
"""		"""
...		...

		@remote_api_endpoint("directory_set_date")
def directory_set_date(self, dates: Dict[Sha1Git, datetime]) -> bool:		def directory_set_date(self, dates: Dict[Sha1Git, datetime]) -> bool:
"""Associate dates to directories identified by sha1 ids, as paired in		"""Associate dates to directories identified by sha1 ids, as paired in
`dates`. Return a boolean stating whether the information was successfully		`dates`. Return a boolean stating whether the information was successfully
stored.		stored.
"""		"""
...		...

		@remote_api_endpoint("directory_get")
def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:		def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
"""Retrieve the associated date for each directory sha1 in `ids`. If some		"""Retrieve the associated date for each directory sha1 in `ids`. If some
directory has no associated date, it is not present in the resulting dictionary.		directory has no associated date, it is not present in the resulting dictionary.
"""		"""
...		...

		@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.		model.
"""		"""
...		...

		@remote_api_endpoint("location_get")
def location_get(self) -> Set[bytes]:		def location_get(self) -> Set[bytes]:
"""Retrieve all paths present in the provenance model."""		"""Retrieve all paths present in the provenance model."""
...		...

		@remote_api_endpoint("origin_set_url")
def origin_set_url(self, urls: Dict[Sha1Git, str]) -> bool:		def origin_set_url(self, urls: Dict[Sha1Git, str]) -> bool:
"""Associate urls to origins identified by sha1 ids, as paired in `urls`. Return		"""Associate urls to origins identified by sha1 ids, as paired in `urls`. Return
a boolean stating whether the information was successfully stored.		a boolean stating whether the information was successfully stored.
"""		"""
...		...

		@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`. If some origin has		"""Retrieve the associated url for each origin sha1 in `ids`. If some origin has
no associated date, it is not present in the resulting dictionary.		no associated date, it is not present in the resulting dictionary.
"""		"""
...		...

		@remote_api_endpoint("revision_set_date")
def revision_set_date(self, dates: Dict[Sha1Git, datetime]) -> bool:		def revision_set_date(self, dates: Dict[Sha1Git, datetime]) -> bool:
"""Associate dates to revisions identified by sha1 ids, as paired in `dates`.		"""Associate dates to revisions identified by sha1 ids, as paired in `dates`.
Return a boolean stating whether the information was successfully stored.		Return a boolean stating whether the information was successfully stored.
"""		"""
...		...

		@remote_api_endpoint("revision_set_origin")
def revision_set_origin(self, origins: Dict[Sha1Git, Sha1Git]) -> bool:		def revision_set_origin(self, origins: Dict[Sha1Git, Sha1Git]) -> bool:
"""Associate origins to revisions identified by sha1 ids, as paired in		"""Associate origins to revisions identified by sha1 ids, as paired in
`origins` (revision ids are keys and origin ids, values). Return a boolean		`origins` (revision ids are keys and origin ids, values). Return a boolean
stating whether the information was successfully stored.		stating whether the information was successfully stored.
"""		"""
...		...

		@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]:
"""Retrieve the associated date and origin for each revision sha1 in `ids`. If		"""Retrieve the associated date and origin for each revision sha1 in `ids`. If
some revision has no associated date nor origin, it is not present in the		some revision has no associated date nor origin, it is not present in the
resulting dictionary.		resulting dictionary.
"""		"""
...		...

		@remote_api_endpoint("relation_add")
def relation_add(		def relation_add(
self, relation: RelationType, data: Iterable[RelationData]		self, relation: RelationType, data: Iterable[RelationData]
) -> bool:		) -> bool:
"""Add entries in the selected `relation`."""		"""Add entries in the selected `relation`."""
...		...

		@remote_api_endpoint("relation_get")
def relation_get(		def relation_get(
self, relation: RelationType, ids: Iterable[Sha1Git], reverse: bool = False		self, relation: RelationType, ids: Iterable[Sha1Git], reverse: bool = False
) -> Set[RelationData]:		) -> Set[RelationData]:
"""Retrieve all entries in the selected `relation` whose source entities are		"""Retrieve all entries in the selected `relation` whose source entities are
identified by some sha1 id in `ids`. If `reverse` is set, destination entities		identified by some sha1 id in `ids`. If `reverse` is set, destination entities
are matched instead.		are matched instead.
"""		"""
...		...

		@remote_api_endpoint("relation_get_all")
def relation_get_all(self, relation: RelationType) -> Set[RelationData]:		def relation_get_all(self, relation: RelationType) -> Set[RelationData]:
"""Retrieve all entries in the selected `relation` that are present in the		"""Retrieve all entries in the selected `relation` that are present in the
provenance model.		provenance model.
"""		"""
...		...

		@remote_api_endpoint("with_path")
		def with_path(self) -> bool:
		...


@runtime_checkable		@runtime_checkable
class ProvenanceInterface(Protocol):		class ProvenanceInterface(Protocol):
storage: ProvenanceStorageInterface		storage: ProvenanceStorageInterface

def flush(self) -> None:		def flush(self) -> None:
"""Flush internal cache to the underlying `storage`."""		"""Flush internal cache to the underlying `storage`."""
...		...
▲ Show 20 Lines • Show All 128 Lines • Show Last 20 Lines