Changeset View
Changeset View
Standalone 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 |