Page MenuHomeSoftware Heritage
Differential D8978 Diff 32383 swh/graphql/resolvers/content.py

Changeset View

Standalone View

View Options

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 Union from typing import Dict, List, Optional, Union
from swh.graphql.errors import 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 .base_connection import BaseList
from .base_node import BaseSWHNode from .base_node import BaseSWHNode
from .directory_entry import BaseDirectoryEntryNode from .directory_entry import BaseDirectoryEntryNode
from .release import BaseReleaseNode from .release import BaseReleaseNode
from .search import SearchResultNode from .search import SearchResultNode
from .snapshot_branch import BaseSnapshotBranchNode from .snapshot_branch import BaseSnapshotBranchNode
vlorentzUnsubmitted
Not Done
Inline Actions
from .snapshot_branch import BaseSnapshotBranchNode
- def read_and_validate_content_hashes(hashes):
+ def read_and_validate_content_hashes(hashes: List[Tuple[str, str]]) -> Dict[str, bytes]:
try:
vlorentz:
def read_and_validate_content_hashes(hashes) -> Dict[str, bytes]:
try:
return {
hash_type: hashutil.hash_to_bytes(hash_value)
for (hash_type, hash_value) in hashes
}
except ValueError as e:
# raise an input error in case of an invalid hash
raise InvalidInputError("Invalid content hash", e)
class BaseContentNode(BaseSWHNode): class BaseContentNode(BaseSWHNode):
""" """
Base resolver for all the content nodes Base resolver for all the content nodes
""" """
def _get_content_by_hashes(self, hashes: dict):
content = self.archive.get_contents(hashes)
# in case of a conflict, return the first element
return content[0] if content else None
@property @property
def hashes(self): def hashes(self):
# FIXME, use a Node instead # FIXME, use a Node instead
return {k: v.hex() for (k, v) in self._node.hashes().items()} return {k: v.hex() for (k, v) in self._node.hashes().items()}
@property @property
def id(self): def id(self):
return self._node.sha1_git return self._node.sha1_git
Show All 24 Lines def license(self):
return None return None
def is_type_of(self): def is_type_of(self):
# is_type_of is required only when resolving a UNION type # is_type_of is required only when resolving a UNION type
# This is for ariadne to return the right type # This is for ariadne to return the right type
return "Content" return "Content"
class ContentNode(BaseContentNode): class ContentbyHashesNode(BaseContentNode):
""" """
Node resolver for a content requested directly with its SWHID Node resolver for a content requested with all of its hashes
A single content object will be returned
""" """
def _get_node_data(self): def _get_node_data(self) -> Optional[Content]:
hashes = {"sha1_git": self.kwargs.get("swhid").object_id} hashes = read_and_validate_content_hashes(self.kwargs.items())
return self._get_content_by_hashes(hashes) contents = self.archive.get_contents(hashes=hashes)
if len(contents) > 1:
# This situation is not expected to happen IRL
class HashContentNode(BaseContentNode): raise DataError("Content hash conflict for the set ", hashes)
""" return contents[0] if contents else None
Node resolver for a content requested with one or more hashes
"""
def _get_node_data(self):
try:
hashes = {
hash_type: hashutil.hash_to_bytes(hash_value)
for (hash_type, hash_value) in self.kwargs.items()
}
except ValueError as e:
# raise an input error in case of an invalid hash
raise InvalidInputError("Invalid content hash", e)
if not hashes:
raise InvalidInputError("At least one of the four hashes must be provided")
return self._get_content_by_hashes(hashes)
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: Union[
SearchResultNode, SearchResultNode,
BaseDirectoryEntryNode, BaseDirectoryEntryNode,
BaseReleaseNode, BaseReleaseNode,
BaseSnapshotBranchNode, BaseSnapshotBranchNode,
] ]
def _get_node_data(self): def _get_node_data(self) -> Optional[Content]:
return self._get_content_by_hashes(hashes={"sha1_git": self.obj.target_hash}) # FIXME, this is not considering hash collisions
# and could return a wrong object in very rare situations
contents = self.archive.get_contents(hashes={"sha1_git": self.obj.target_hash})
# always returning the first content from the storage
return contents[0] if contents else None
class ContentSwhidList(BaseList):
"""
Return a non paginated list of contents for the given SWHID
This will return a single item in most of the cases
"""
_node_class = BaseContentNode
vlorentzUnsubmitted
Not Done
Inline Actions

Could you add type annotations to this function and the ones below?

vlorentz: Could you add type annotations to this function and the ones below?
def _get_results(self) -> List[Content]:
hashes = {"sha1_git": self.kwargs.get("swhid").object_id}
return self.archive.get_contents(hashes=hashes)
class ContentHashList(BaseList):
"""
Return a non paginated list of contents for the given hashes
This will return a single item in most of the cases
"""
_node_class = BaseContentNode
def _get_results(self) -> List[Content]:
hashes = read_and_validate_content_hashes(self.kwargs.items())
if not hashes:
raise InvalidInputError("At least one of the four hashes must be provided")
return self.archive.get_contents(hashes=hashes)
About · Developer information · Legal