Differential D6307 Diff 23959 swh/provenance/mongo/backend_normalized.py

Changeset View

Standalone View

swh/provenance/mongo/backend_normalized.py

This file was added.

				# Copyright (C) 2021 The Software Heritage developers
				# See the AUTHORS file at the top-level directory of this distribution
				# License: GNU General Public License version 3, or any later version
				# See top-level LICENSE file for more information

				from .backend import ProvenanceStorageMongoDb

				class MongoNormalized(ProvenanceStorageMongoDb):

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_add"})
				def content_add(
				self, cnts: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
				) -> bool:
				return True

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_add"})
				def directory_add(
				self, dirs: Union[Iterable[Sha1Git], Dict[Sha1Git, Optional[datetime]]]
				) -> bool:
				return True

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_first"})
				def content_find_first(self, id: Sha1Git) -> Optional[ProvenanceResult]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_find_all"})
				def content_find_all(
				self, id: Sha1Git, limit: Optional[int] = None
				) -> Generator[ProvenanceResult, None, None]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "content_get"})
				def content_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "directory_get"})
				def directory_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, datetime]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "entity_get_all"})
				def entity_get_all(self, entity: EntityType) -> Set[Sha1Git]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_add"})
				def location_add(self, paths: Iterable[bytes]) -> bool:
				# TODO: implement this methods if path are to be stored in a separate collection
				return True

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "location_get_all"})
				def location_get_all(self) -> Set[bytes]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_add"})
				def origin_add(self, orgs: Dict[Sha1Git, str]) -> bool:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "origin_get"})
				def origin_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, str]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_add"})
				def revision_add(
				self, revs: Union[Iterable[Sha1Git], Dict[Sha1Git, RevisionData]]
				) -> bool:
				return True

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "revision_get"})
				def revision_get(self, ids: Iterable[Sha1Git]) -> Dict[Sha1Git, RevisionData]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_add"})
				def relation_add(
				self, relation: RelationType, data: Dict[Sha1Git, Set[RelationData]]
				) -> bool:
				return True

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get"})
				def relation_get(
				self, relation: RelationType, ids: Iterable[Sha1Git], reverse: bool = False
				) -> Dict[Sha1Git, Set[RelationData]]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "relation_get_all"})
				def relation_get_all(
				self, relation: RelationType
				) -> Dict[Sha1Git, Set[RelationData]]:
				pass

				@statsd.timed(metric=STORAGE_DURATION_METRIC, tags={"method": "with_path"})
				def with_path(self) -> bool:
				return True