-
Notifications
You must be signed in to change notification settings - Fork 9
Expand file tree
/
Copy pathcomputers.py
More file actions
74 lines (58 loc) · 2 KB
/
computers.py
File metadata and controls
74 lines (58 loc) · 2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
"""Defines plugins for AiiDA computers."""
# pylint: disable=too-few-public-methods,redefined-builtin,,unused-argument
from typing import Any, Optional
import graphene as gr
from aiida.orm import Computer
from aiida_restapi.graphql.filter_syntax import parse_filter_str
from aiida_restapi.graphql.plugins import QueryPlugin
from .nodes import NodesQuery
from .orm_factories import (
ENTITY_DICT_TYPE,
multirow_cls_factory,
resolve_entity,
single_cls_factory,
)
from .utils import FilterString
class ComputerQuery(single_cls_factory(Computer)): # type: ignore[misc]
"""Query an AiiDA Computer"""
nodes = gr.Field(NodesQuery, filters=FilterString())
@staticmethod
def resolve_nodes(parent: Any, info: gr.ResolveInfo, filters: Optional[str] = None) -> dict:
"""Resolution function."""
# pass filter specification to NodesQuery
parsed_filters = parse_filter_str(filters)
parsed_filters['dbcomputer_id'] = parent['id']
return {'filters': parsed_filters}
class ComputersQuery(multirow_cls_factory(ComputerQuery, Computer, 'computers')): # type: ignore[misc]
"""Query all AiiDA Computers"""
def resolve_Computer(
parent: Any,
info: gr.ResolveInfo,
id: Optional[int] = None,
uuid: Optional[str] = None,
) -> ENTITY_DICT_TYPE:
"""Resolution function."""
return resolve_entity(Computer, info, id, uuid)
def resolve_Computers(parent: Any, info: gr.ResolveInfo, filters: Optional[str] = None) -> dict:
"""Resolution function."""
# pass filter to ComputersQuery
return {'filters': parse_filter_str(filters)}
ComputerQueryPlugin = QueryPlugin(
'computer',
gr.Field(
ComputerQuery,
id=gr.Int(),
uuid=gr.String(),
description='Query for a single Computer',
),
resolve_Computer,
)
ComputersQueryPlugin = QueryPlugin(
'computers',
gr.Field(
ComputersQuery,
description='Query for multiple Computers',
filters=FilterString(),
),
resolve_Computers,
)