Changeset View
Changeset View
Standalone View
Standalone View
swh/graphql/resolvers/resolvers.py
| Show All 10 Lines | |||||
| # and in the following priority order | # and in the following priority order | ||||
| # - In this module using a decorator (eg: @visitstatus.field("snapshot") | # - In this module using a decorator (eg: @visitstatus.field("snapshot") | ||||
| # Every object (type) is expected to resolve this way as they can accept arguments | # Every object (type) is expected to resolve this way as they can accept arguments | ||||
| # eg: origin.visits takes arguments to paginate | # eg: origin.visits takes arguments to paginate | ||||
| # - As a property in the Node object (eg: resolvers.visit.BaseVisitNode.id) | # - As a property in the Node object (eg: resolvers.visit.BaseVisitNode.id) | ||||
| # Every scalar is expected to resolve this way | # Every scalar is expected to resolve this way | ||||
| # - As an attribute/item in the object/dict returned by a backend (eg: Origin.url) | # - As an attribute/item in the object/dict returned by a backend (eg: Origin.url) | ||||
| from typing import Optional | from typing import Optional, Union | ||||
| from ariadne import ObjectType, UnionType | from ariadne import ObjectType, UnionType | ||||
| from graphql.type import GraphQLResolveInfo | from graphql.type import GraphQLResolveInfo | ||||
| from swh.graphql import resolvers as rs | from swh.graphql import resolvers as rs | ||||
| from swh.graphql.utils import utils | from swh.graphql.utils import utils | ||||
| from .resolver_factory import get_connection_resolver, get_node_resolver | from .resolver_factory import get_connection_resolver, get_node_resolver | ||||
| Show All 24 Lines | |||||
| def origin_resolver(obj: None, info: GraphQLResolveInfo, **kw) -> rs.origin.OriginNode: | def origin_resolver(obj: None, info: GraphQLResolveInfo, **kw) -> rs.origin.OriginNode: | ||||
| """ """ | """ """ | ||||
| resolver = get_node_resolver("origin") | resolver = get_node_resolver("origin") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @origin.field("latestVisit") | @origin.field("latestVisit") | ||||
| def latest_visit_resolver( | def latest_visit_resolver( | ||||
| obj: rs.origin.OriginNode, info: GraphQLResolveInfo, **kw | obj: rs.origin.BaseOriginNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.visit.LatestVisitNode: | ) -> rs.visit.LatestVisitNode: | ||||
| """ """ | """ """ | ||||
| resolver = get_node_resolver("latest-visit") | resolver = get_node_resolver("latest-visit") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("visit") | @query.field("visit") | ||||
| def visit_resolver( | def visit_resolver( | ||||
| obj: None, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.visit.OriginVisitNode: | ) -> rs.visit.OriginVisitNode: | ||||
| """ """ | """ """ | ||||
| resolver = get_node_resolver("visit") | resolver = get_node_resolver("visit") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @visit.field("latestStatus") | @visit.field("latestStatus") | ||||
| def latest_visit_status_resolver( | def latest_visit_status_resolver( | ||||
| obj, info: GraphQLResolveInfo, **kw | obj: rs.visit.BaseVisitNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.visit_status.LatestVisitStatusNode: | ) -> rs.visit_status.LatestVisitStatusNode: | ||||
| """ """ | """ """ | ||||
| resolver = get_node_resolver("latest-status") | resolver = get_node_resolver("latest-status") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("snapshot") | @query.field("snapshot") | ||||
| def snapshot_resolver( | def snapshot_resolver( | ||||
| Show All 12 Lines | if obj.snapshotSWHID is None: | ||||
| return None | return None | ||||
| resolver = get_node_resolver("visit-snapshot") | resolver = get_node_resolver("visit-snapshot") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @snapshot_branch.field("target") | @snapshot_branch.field("target") | ||||
| def snapshot_branch_target_resolver( | def snapshot_branch_target_resolver( | ||||
| obj: rs.snapshot_branch.BaseSnapshotBranchNode, info: GraphQLResolveInfo, **kw | obj: rs.snapshot_branch.BaseSnapshotBranchNode, info: GraphQLResolveInfo, **kw | ||||
| ): | ) -> Union[ | ||||
| rs.revision.BaseRevisionNode, | |||||
| rs.release.BaseReleaseNode, | |||||
| rs.directory.BaseDirectoryNode, | |||||
| rs.content.BaseContentNode, | |||||
| rs.snapshot.BaseSnapshotNode, | |||||
| rs.snapshot_branch.BaseSnapshotBranchNode, | |||||
| ]: | |||||
| """ | """ | ||||
| Snapshot branch target can be a revision, release, directory, | Snapshot branch target can be a revision, release, directory, | ||||
| content, snapshot or a branch itself (alias type) | content, snapshot or a branch itself (alias type) | ||||
| """ | """ | ||||
| resolver_type = f"branch-{obj.type}" | resolver_type = f"branch-{obj.type}" | ||||
| resolver = get_node_resolver(resolver_type) | resolver = get_node_resolver(resolver_type) | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("revision") | @query.field("revision") | ||||
| def revision_resolver( | def revision_resolver( | ||||
| obj: None, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.revision.RevisionNode: | ) -> rs.revision.RevisionNode: | ||||
| resolver = get_node_resolver("revision") | resolver = get_node_resolver("revision") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @revision.field("directory") | @revision.field("directory") | ||||
| def revision_directory_resolver( | def revision_directory_resolver( | ||||
| obj, info: GraphQLResolveInfo, **kw | obj: rs.revision.BaseRevisionNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.directory.RevisionDirectoryNode: | ) -> rs.directory.RevisionDirectoryNode: | ||||
| resolver = get_node_resolver("revision-directory") | resolver = get_node_resolver("revision-directory") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("release") | @query.field("release") | ||||
| def release_resolver( | def release_resolver( | ||||
| obj: None, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.release.ReleaseNode: | ) -> rs.release.ReleaseNode: | ||||
| resolver = get_node_resolver("release") | resolver = get_node_resolver("release") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @release.field("target") | @release.field("target") | ||||
| def release_target_resolver(obj, info: GraphQLResolveInfo, **kw): | def release_target_resolver( | ||||
| obj: rs.release.BaseReleaseNode, info: GraphQLResolveInfo, **kw | |||||
| ) -> Union[ | |||||
| rs.revision.BaseRevisionNode, | |||||
| rs.release.BaseReleaseNode, | |||||
| rs.directory.BaseDirectoryNode, | |||||
| rs.content.BaseContentNode, | |||||
| ]: | |||||
| """ | """ | ||||
| release target can be a release, revision, | release target can be a release, revision, | ||||
| directory or content | directory or content | ||||
| obj is release here, target type is | obj is release here, target type is | ||||
| obj.target_type | obj.target_type | ||||
| """ | """ | ||||
| resolver_type = f"release-{obj.target_type.value}" | resolver_type = f"release-{obj.target_type.value}" | ||||
| resolver = get_node_resolver(resolver_type) | resolver = get_node_resolver(resolver_type) | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("directory") | @query.field("directory") | ||||
| def directory_resolver( | def directory_resolver( | ||||
| obj: None, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.directory.DirectoryNode: | ) -> rs.directory.DirectoryNode: | ||||
| resolver = get_node_resolver("directory") | resolver = get_node_resolver("directory") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("directoryEntry") | @query.field("directoryEntry") | ||||
| def directory_entry_resolver( | def directory_entry_resolver( | ||||
| obj: None, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.directory.DirectoryNode: | ) -> rs.directory_entry.DirectoryEntryNode: | ||||
| resolver = get_node_resolver("directory-entry") | resolver = get_node_resolver("directory-entry") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @directory_entry.field("target") | @directory_entry.field("target") | ||||
| def directory_entry_target_resolver( | def directory_entry_target_resolver( | ||||
| obj: rs.directory_entry.BaseDirectoryEntryNode, info: GraphQLResolveInfo, **kw | obj: rs.directory_entry.BaseDirectoryEntryNode, info: GraphQLResolveInfo, **kw | ||||
| ): | ) -> Union[ | ||||
| rs.revision.BaseRevisionNode, | |||||
| rs.directory.BaseDirectoryNode, | |||||
| rs.content.BaseContentNode, | |||||
| ]: | |||||
| """ | """ | ||||
| directory entry target can be a directory or a content | directory entry target can be a directory, content or a revision | ||||
| """ | """ | ||||
| resolver_type = f"dir-entry-{obj.type}" | resolver_type = f"dir-entry-{obj.type}" | ||||
| resolver = get_node_resolver(resolver_type) | resolver = get_node_resolver(resolver_type) | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("content") | @query.field("content") | ||||
| def content_resolver( | def content_resolver( | ||||
| obj: None, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.content.ContentNode: | ) -> rs.content.ContentNode: | ||||
| resolver = get_node_resolver("content") | resolver = get_node_resolver("content") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @search_result.field("target") | @search_result.field("target") | ||||
| def search_result_target_resolver( | def search_result_target_resolver( | ||||
| obj: rs.search.SearchResultNode, info: GraphQLResolveInfo, **kw | obj: rs.search.SearchResultNode, info: GraphQLResolveInfo, **kw | ||||
| ): | ) -> Union[ | ||||
| rs.origin.BaseOriginNode, | |||||
| rs.snapshot.BaseSnapshotNode, | |||||
| rs.revision.BaseRevisionNode, | |||||
| rs.release.BaseReleaseNode, | |||||
| rs.directory.BaseDirectoryNode, | |||||
| rs.content.BaseContentNode, | |||||
| ]: | |||||
| resolver_type = f"search-result-{obj.type}" | resolver_type = f"search-result-{obj.type}" | ||||
| resolver = get_node_resolver(resolver_type) | resolver = get_node_resolver(resolver_type) | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("contentByHash") | @query.field("contentByHash") | ||||
| def content_by_hash_resolver( | def content_by_hash_resolver( | ||||
| obj: None, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| Show All 11 Lines | def origins_resolver( | ||||
| obj: None, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.origin.OriginConnection: | ) -> rs.origin.OriginConnection: | ||||
| resolver = get_connection_resolver("origins") | resolver = get_connection_resolver("origins") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @origin.field("visits") | @origin.field("visits") | ||||
| def visits_resolver( | def visits_resolver( | ||||
| obj: rs.origin.OriginNode, info: GraphQLResolveInfo, **kw | obj: rs.origin.BaseOriginNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.visit.OriginVisitConnection: | ) -> rs.visit.OriginVisitConnection: | ||||
| resolver = get_connection_resolver("origin-visits") | resolver = get_connection_resolver("origin-visits") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @origin.field("snapshots") | @origin.field("snapshots") | ||||
| def origin_snapshots_resolver( | def origin_snapshots_resolver( | ||||
| obj: rs.origin.OriginNode, info: GraphQLResolveInfo, **kw | obj: rs.origin.BaseOriginNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.snapshot.OriginSnapshotConnection: | ) -> rs.snapshot.OriginSnapshotConnection: | ||||
| """ """ | """ """ | ||||
| resolver = get_connection_resolver("origin-snapshots") | resolver = get_connection_resolver("origin-snapshots") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @visit.field("status") | @visit.field("status") | ||||
| def visitstatus_resolver( | def visitstatus_resolver( | ||||
| obj, info: GraphQLResolveInfo, **kw | obj: rs.visit.BaseVisitNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.visit_status.VisitStatusConnection: | ) -> rs.visit_status.VisitStatusConnection: | ||||
| resolver = get_connection_resolver("visit-status") | resolver = get_connection_resolver("visit-status") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @snapshot.field("branches") | @snapshot.field("branches") | ||||
| def snapshot_branches_resolver( | def snapshot_branches_resolver( | ||||
| obj, info: GraphQLResolveInfo, **kw | obj: rs.snapshot.BaseSnapshotNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.snapshot_branch.SnapshotBranchConnection: | ) -> rs.snapshot_branch.SnapshotBranchConnection: | ||||
| resolver = get_connection_resolver("snapshot-branches") | resolver = get_connection_resolver("snapshot-branches") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @revision.field("parents") | @revision.field("parents") | ||||
| def revision_parents_resolver( | def revision_parents_resolver( | ||||
| obj, info: GraphQLResolveInfo, **kw | obj: rs.revision.BaseRevisionNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.revision.ParentRevisionConnection: | ) -> rs.revision.ParentRevisionConnection: | ||||
| resolver = get_connection_resolver("revision-parents") | resolver = get_connection_resolver("revision-parents") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @revision.field("revisionLog") | @revision.field("revisionLog") | ||||
| def revision_log_resolver(obj, info, **kw): | def revision_log_resolver( | ||||
| obj: rs.revision.BaseRevisionNode, info: GraphQLResolveInfo, **kw | |||||
anlambert: `-> rs.revision.LogRevisionConnection` | |||||
| ) -> rs.revision.LogRevisionConnection: | |||||
| resolver = get_connection_resolver("revision-log") | resolver = get_connection_resolver("revision-log") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @directory.field("entries") | @directory.field("entries") | ||||
| def directory_entries_resolver( | def directory_entries_resolver( | ||||
| obj, info: GraphQLResolveInfo, **kw | obj: rs.directory.BaseDirectoryNode, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.directory_entry.DirectoryEntryConnection: | ) -> rs.directory_entry.DirectoryEntryConnection: | ||||
| resolver = get_connection_resolver("directory-entries") | resolver = get_connection_resolver("directory-entries") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("resolveSwhid") | @query.field("resolveSwhid") | ||||
| def search_swhid_resolver( | def search_swhid_resolver( | ||||
| obj, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.search.ResolveSwhidConnection: | ) -> rs.search.ResolveSwhidConnection: | ||||
| resolver = get_connection_resolver("resolve-swhid") | resolver = get_connection_resolver("resolve-swhid") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| @query.field("search") | @query.field("search") | ||||
| def search_resolver( | def search_resolver( | ||||
| obj, info: GraphQLResolveInfo, **kw | obj: None, info: GraphQLResolveInfo, **kw | ||||
| ) -> rs.search.ResolveSwhidConnection: | ) -> rs.search.SearchConnection: | ||||
| resolver = get_connection_resolver("search") | resolver = get_connection_resolver("search") | ||||
| return resolver(obj, info, **kw) | return resolver(obj, info, **kw) | ||||
| # Any other type of resolver | # Any other type of resolver | ||||
| @release_target.type_resolver | @release_target.type_resolver | ||||
| @directory_entry_target.type_resolver | @directory_entry_target.type_resolver | ||||
| @branch_target.type_resolver | @branch_target.type_resolver | ||||
| @search_result_target.type_resolver | @search_result_target.type_resolver | ||||
| def union_resolver(obj, *_) -> str: | def union_resolver( | ||||
| obj: Union[ | |||||
| rs.origin.BaseOriginNode, | |||||
| rs.revision.BaseRevisionNode, | |||||
| rs.release.BaseReleaseNode, | |||||
| rs.directory.BaseDirectoryNode, | |||||
| rs.content.BaseContentNode, | |||||
| rs.snapshot.BaseSnapshotNode, | |||||
| rs.snapshot_branch.BaseSnapshotBranchNode, | |||||
| ], | |||||
| *_, | |||||
| ) -> str: | |||||
| """ | """ | ||||
| Generic resolver for all the union types | Generic resolver for all the union types | ||||
| """ | """ | ||||
| return obj.is_type_of() | return obj.is_type_of() | ||||
| @binary_string.field("text") | @binary_string.field("text") | ||||
| def binary_string_text_resolver(obj, *args, **kw): | def binary_string_text_resolver(obj: bytes, *args, **kw) -> str: | ||||
| return obj.decode(utils.ENCODING, "replace") | return obj.decode(utils.ENCODING, "replace") | ||||
| @binary_string.field("base64") | @binary_string.field("base64") | ||||
| def binary_string_base64_resolver(obj, *args, **kw): | def binary_string_base64_resolver(obj: bytes, *args, **kw) -> str: | ||||
| return utils.get_b64_string(obj) | return utils.get_b64_string(obj) | ||||
-> rs.revision.LogRevisionConnection