Changeset View
Changeset View
Standalone View
Standalone View
swh/graphql/resolvers/snapshot_branch.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 collections import namedtuple | from collections import namedtuple | ||||
| from swh.graphql.errors import ObjectNotFoundError | from swh.graphql.errors import ObjectNotFoundError | ||||
| from swh.graphql.utils import utils | from swh.graphql.utils import utils | ||||
| from swh.storage.interface import PagedResult | from swh.storage.interface import PagedResult | ||||
| from .base_connection import BaseConnection | from .base_connection import BaseConnection | ||||
| from .base_node import BaseNode | from .base_node import BaseNode | ||||
| from .target import TargetNode | |||||
| class BaseSnapshotBranchNode(BaseNode): | class BaseSnapshotBranchNode(BaseNode): | ||||
| # target field for this node is a UNION type | # target field for this node is a UNION type | ||||
| # It is resolved in the top level (resolvers.resolvers.py) | # It is resolved in the top level (resolvers.resolvers.py) | ||||
| def _get_node_from_data(self, node_data: tuple): | def _get_node_from_data(self, node_data: tuple): | ||||
| # node_data is a tuple as returned by _get_paged_result in | # node_data is a tuple as returned by _get_paged_result in | ||||
| # SnapshotBranchConnection and _get_node_data in AliasSnapshotBranchNode | # SnapshotBranchConnection and _get_node_data in AliasSnapshotBranchNode | ||||
| # overriding to support this special data structure | # overriding to support this special data structure | ||||
| branch_name, branch_obj = node_data | branch_name, branch_obj = node_data | ||||
| node = { | node = { | ||||
| "name": branch_name, | "name": branch_name, | ||||
| "type": branch_obj.target_type.value, | "target_type": branch_obj.target_type.value, | ||||
| "target_hash": branch_obj.target, | "target_id": branch_obj.target, | ||||
| } | } | ||||
| return namedtuple("NodeObj", node.keys())(*node.values()) | return namedtuple("NodeObj", node.keys())(*node.values()) | ||||
| def is_type_of(self): | def is_type_of(self): | ||||
| return "Branch" | return "Branch" | ||||
| @property | |||||
| def targetType(self): # To support the schema naming convention | |||||
| return self._node.type | |||||
| def snapshot_swhid(self): | def snapshot_swhid(self): | ||||
| """ | """ | ||||
| Logic to handle multiple branch alias redirects | Logic to handle multiple branch alias redirects | ||||
| Alias redirects can be any level deep. Hence the parent snapshot can be | Alias redirects can be any level deep. Hence the parent snapshot can be | ||||
| reached only by a loop | reached only by a loop | ||||
| This code expects every BranchNode to have a Snapshot parent in the GraphQL query | This code expects every BranchNode to have a Snapshot parent in the GraphQL query | ||||
| at some level. | at some level. | ||||
| """ | """ | ||||
| from .snapshot import BaseSnapshotNode | from .snapshot import BaseSnapshotNode | ||||
| parent = self.obj | parent = self.obj | ||||
| while parent: | while parent: | ||||
| if isinstance( | if isinstance( | ||||
| parent, BaseSnapshotNode | parent, BaseSnapshotNode | ||||
| ): # Reached the nearest SnapshotNode object | ): # Reached the nearest SnapshotNode object | ||||
| return parent.swhid | return parent.swhid | ||||
| parent = parent.obj | parent = parent.obj | ||||
| # Reached the root query node. This will never happen with the current entrypoints | # Reached the root query node. This will never happen with the current entrypoints | ||||
| raise ObjectNotFoundError("There is no snapshot associated with the branch") | raise ObjectNotFoundError("There is no snapshot associated with the branch") | ||||
| class AliasSnapshotBranchNode(BaseSnapshotBranchNode): | class TargetSnapshotBranchNode(BaseSnapshotBranchNode): | ||||
| obj: BaseSnapshotBranchNode | obj: TargetNode | ||||
| def _get_node_data(self): | def _get_node_data(self): | ||||
| snapshot_swhid = self.snapshot_swhid() | snapshot_swhid = self.snapshot_swhid() | ||||
| target_branch = self.obj.target_hash | target_branch = self.obj.target_id | ||||
| alias_branch = self.archive.get_snapshot_branches( | alias_branch = self.archive.get_snapshot_branches( | ||||
| snapshot_swhid.object_id, first=1, name_include=target_branch | snapshot_swhid.object_id, first=1, name_include=target_branch | ||||
| ) | ) | ||||
| if target_branch not in alias_branch["branches"]: | if target_branch not in alias_branch["branches"]: | ||||
| raise ObjectNotFoundError( | raise ObjectNotFoundError( | ||||
| f"Branch name with {target_branch.decode()} is not available" | f"Branch name with {target_branch.decode()} is not available" | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 53 Lines • Show Last 20 Lines | |||||