Changeset View
Changeset View
Standalone View
Standalone View
swh/graphql/backends/archive.py
| # Copyright (C) 2022 The Software Heritage developers | # Copyright (C) 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 | ||||
| import os | |||||
| from typing import Any, Dict, Optional | |||||
| from swh.graphql import server | from swh.graphql import server | ||||
| from swh.model.model import Sha1Git | |||||
| from swh.model.swhids import ObjectType | from swh.model.swhids import ObjectType | ||||
| class Archive: | class Archive: | ||||
| def __init__(self): | def __init__(self): | ||||
| self.storage = server.get_storage() | self.storage = server.get_storage() | ||||
| def get_origin(self, url): | def get_origin(self, url): | ||||
| Show All 37 Lines | def get_revisions(self, revision_ids): | ||||
| return self.storage.revision_get(revision_ids=revision_ids) | return self.storage.revision_get(revision_ids=revision_ids) | ||||
| def get_revision_log(self, revision_ids, after=None, first=50): | def get_revision_log(self, revision_ids, after=None, first=50): | ||||
| return self.storage.revision_log(revisions=revision_ids, limit=first) | return self.storage.revision_log(revisions=revision_ids, limit=first) | ||||
| def get_releases(self, release_ids): | def get_releases(self, release_ids): | ||||
| return self.storage.release_get(releases=release_ids) | return self.storage.release_get(releases=release_ids) | ||||
| def get_directory_entry_by_path( | |||||
| self, directory_id: Sha1Git, path: str | |||||
| ) -> Optional[Dict[str, Any]]: | |||||
| paths = [x.encode() for x in path.strip(os.path.sep).split(os.path.sep)] | |||||
| return self.storage.directory_entry_get_by_path( | |||||
| directory=directory_id, paths=paths | |||||
| ) | |||||
| def get_directory_entries(self, directory_id, after=None, first=50): | def get_directory_entries(self, directory_id, after=None, first=50): | ||||
| return self.storage.directory_get_entries( | return self.storage.directory_get_entries( | ||||
| directory_id, limit=first, page_token=after | directory_id, limit=first, page_token=after | ||||
| ) | ) | ||||
| def is_object_available(self, object_id: str, object_type: ObjectType) -> bool: | def is_object_available(self, object_id: str, object_type: ObjectType) -> bool: | ||||
| mapping = { | mapping = { | ||||
| ObjectType.CONTENT: self.storage.content_missing_per_sha1_git, | ObjectType.CONTENT: self.storage.content_missing_per_sha1_git, | ||||
| Show All 12 Lines | |||||