Differential D6712 Diff 24725 swh/provenance/interface.py

Changeset View

Standalone View

swh/provenance/interface.py

Show All 38 Lines	class ProvenanceResult:
content: Sha1Git		content: Sha1Git
revision: Sha1Git		revision: Sha1Git
date: datetime		date: datetime
origin: Optional[str]		origin: Optional[str]
path: bytes		path: bytes


@dataclass(eq=True, frozen=True)		@dataclass(eq=True, frozen=True)
		class DirectoryData:
		"""Object representing the data associated to a directory in the provenance model,
		where `date` is the date of the directory in the isochrone frontier, and `flat` is a
		flag acknowledging that a flat model for the elements outside the frontier has
		already been created.
		"""

		date: datetime
		flat: bool


		@dataclass(eq=True, frozen=True)
class RevisionData:		class RevisionData:
"""Object representing the data associated to a revision in the provenance model,		"""Object representing the data associated to a revision in the provenance model,
where `date` is the optional date of the revision (specifying it acknowledges that		where `date` is the optional date of the revision (specifying it acknowledges that
the revision was already processed by the revision-content algorithm); and `origin`		the revision was already processed by the revision-content algorithm); and `origin`
identifies the preferred origin for the revision, if any.		identifies the preferred origin for the revision, if any.
"""		"""

date: Optional[datetime]		date: Optional[datetime]
Show All 25 Lines	) -> None:
...		...

@remote_api_endpoint("close")		@remote_api_endpoint("close")
def close(self) -> None:		def close(self) -> None:
"""Close connection to the storage and release resources."""		"""Close connection to the storage and release resources."""
...		...

@remote_api_endpoint("content_add")		@remote_api_endpoint("content_add")
def content_add(		def content_add(self, cnts: Dict[Sha1Git, datetime]) -> bool:
self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]		"""Add blobs identified by sha1 ids, with an associated date (as paired in
) -> bool:		`cnts`) to the provenance storage. Return a boolean stating whether the
"""Add blobs identified by sha1 ids, with an optional associated date (as paired
in `cnts`) to the provenance storage. Return a boolean stating whether the
information was successfully stored.		information was successfully stored.
"""		"""
...		...

@remote_api_endpoint("content_find_first")		@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")		@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_get")		@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`."""
no associated date, it is not present in the resulting dictionary.
"""
...		...

@remote_api_endpoint("directory_add")		@remote_api_endpoint("directory_add")
def directory_add(		def directory_add(self, dirs: Dict[Sha1Git, DirectoryData]) -> bool:
self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]		"""Add directories identified by sha1 ids, with associated date and (optional)
) -> bool:		flatten flag (as paired in `dirs`) to the provenance storage. If the flatten
"""Add directories identified by sha1 ids, with an optional associated date (as		flag is set to None, the previous value present in the storage is preserved.
paired in `dirs`) to the provenance storage. Return a boolean stating if the		Return a boolean stating if the information was successfully stored.
information was successfully stored.
"""		"""
...		...

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

@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.
"""		"""
▲ Show 20 Lines • Show All 239 Lines • Show Last 20 Lines