Changeset View
Changeset View
Standalone View
Standalone View
swh/graphql/resolvers/content.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 | ||||
| from typing import Dict, List, Optional, Union | from typing import Dict, List, Optional | ||||
| from swh.graphql.errors import DataError, InvalidInputError | from swh.graphql.errors import DataError, InvalidInputError | ||||
| from swh.model import hashutil | from swh.model import hashutil | ||||
| from swh.model.model import Content | from swh.model.model import Content | ||||
| from .base_connection import BaseList | from .base_connection import BaseList | ||||
| from .base_node import BaseSWHNode | from .base_node import BaseSWHNode | ||||
| from .directory_entry import BaseDirectoryEntryNode | from .target import TargetNode | ||||
| from .release import BaseReleaseNode | |||||
| from .search import SearchResultNode | |||||
| from .snapshot_branch import BaseSnapshotBranchNode | |||||
| def read_and_validate_content_hashes(hashes) -> Dict[str, bytes]: | def read_and_validate_content_hashes(hashes) -> Dict[str, bytes]: | ||||
| try: | try: | ||||
| return { | return { | ||||
| hash_type: hashutil.hash_to_bytes(hash_value) | hash_type: hashutil.hash_to_bytes(hash_value) | ||||
| for (hash_type, hash_value) in hashes | for (hash_type, hash_value) in hashes | ||||
| } | } | ||||
| ▲ Show 20 Lines • Show All 63 Lines • ▼ Show 20 Lines | |||||
| class TargetContentNode(BaseContentNode): | class TargetContentNode(BaseContentNode): | ||||
| """ | """ | ||||
| Node resolver for a content requested as a target | Node resolver for a content requested as a target | ||||
| """ | """ | ||||
| _can_be_null = True | _can_be_null = True | ||||
| obj: Union[ | obj: TargetNode | ||||
| SearchResultNode, | |||||
| BaseDirectoryEntryNode, | |||||
| BaseReleaseNode, | |||||
| BaseSnapshotBranchNode, | |||||
| ] | |||||
| def _get_node_data(self) -> Optional[Content]: | def _get_node_data(self) -> Optional[Content]: | ||||
| # FIXME, this is not considering hash collisions | # FIXME, this is not considering hash collisions | ||||
| # and could return a wrong object in very rare situations | # and could return a wrong object in very rare situations | ||||
| contents = self.archive.get_contents(hashes={"sha1_git": self.obj.target_hash}) | contents = self.archive.get_contents(hashes={"sha1_git": self.obj.target_id}) | ||||
| # always returning the first content from the storage | # always returning the first content from the storage | ||||
| return contents[0] if contents else None | return contents[0] if contents else None | ||||
| class ContentSwhidList(BaseList): | class ContentSwhidList(BaseList): | ||||
| """ | """ | ||||
| Return a non paginated list of contents for the given SWHID | Return a non paginated list of contents for the given SWHID | ||||
| This will return a single item in most of the cases | This will return a single item in most of the cases | ||||
| Show All 22 Lines | |||||