1import asyncio
2
3from typing import Any, Dict, Optional, Union
4
5from tartiflette.coercers.common import CoercionResult, Path
6from tartiflette.coercers.literals.null_and_variable_coercer import (
7    null_and_variable_coercer_wrapper,
8)
9from tartiflette.coercers.literals.utils import is_missing_variable
10from tartiflette.constants import UNDEFINED_VALUE
11from tartiflette.language.ast import ObjectValueNode
12from tartiflette.utils.values import is_invalid_value
13
14__all__ = ("input_object_coercer",)
15
16SKIP_FIELD = object()
17
18
19async def input_field_value_coercer(
20    input_field: "GraphQLInputField",
21    parent_node: Union["VariableDefinitionNode", "InputValueDefinitionNode"],
22    value_node: Union["ValueNode", "VariableNode", "UNDEFINED_VALUE"],
23    ctx: Optional[Any],
24    variables: Optional[Dict[str, Any]],
25    path: Optional["Path"],
26) -> Union["CoercionResult", "UNDEFINED_VALUE", "SKIP_FIELD"]:
27    """
28    Computes the value of an input field.
29    :param input_field: the input field to compute
30    :param parent_node: the root parent AST node
31    :param value_node: the value node to compute
32    :param ctx: context passed to the query execution
33    :param variables: the variables provided in the GraphQL request
34    :param path: the path traveled until this coercer
35    :type input_field: GraphQLInputField
36    :type parent_node: Union[VariableDefinitionNode, InputValueDefinitionNode]
37    :type value_node: Union[ValueNode, VariableNode, UNDEFINED_VALUE]
38    :type ctx: Optional[Any]
39    :type variables: Optional[Dict[str, Any]]
40    :type path: Optional[Path]
41    :return: the computed value
42    :rtype: Union[CoercionResult, UNDEFINED_VALUE, SKIP_FIELD]
43    """
44    if is_invalid_value(value_node) or is_missing_variable(
45        value_node.value, variables
46    ):
47        if input_field.default_value is not None:
48            input_field_node = input_field.default_value
49        elif input_field.graphql_type.is_non_null_type:
50            return UNDEFINED_VALUE
51        else:
52            return SKIP_FIELD
53    else:
54        input_field_node = value_node.value
55
56    return await input_field.literal_coercer(
57        parent_node, input_field_node, ctx, variables=variables, path=path
58    )
59
60
61@null_and_variable_coercer_wrapper
62async def input_object_coercer(
63    parent_node: Union["VariableDefinitionNode", "InputValueDefinitionNode"],
64    node: Union["ValueNode", "VariableNode"],
65    ctx: Optional[Any],
66    input_object_type: "GraphQLInputObjectType",
67    variables: Optional[Dict[str, Any]] = None,
68    path: Optional["Path"] = None,
69) -> "CoercionResult":
70    """
71    Computes the value of an input object.
72    :param parent_node: the root parent AST node
73    :param node: the AST node to treat
74    :param ctx: context passed to the query execution
75    :param input_object_type: the GraphQLInputObjectType instance of the input
76    object
77    :param variables: the variables provided in the GraphQL request
78    :param path: the path traveled until this coercer
79    :type parent_node: Union[VariableDefinitionNode, InputValueDefinitionNode]
80    :type node: Union[ValueNode, VariableNode]
81    :type ctx: Optional[Any]
82    :type input_object_type: GraphQLInputObjectType
83    :type variables: Optional[Dict[str, Any]]
84    :type path: Optional[Path]
85    :return: the computed value
86    :rtype: CoercionResult
87    """
88    # pylint: disable=too-many-locals
89    if not isinstance(node, ObjectValueNode):
90        return CoercionResult(value=UNDEFINED_VALUE)
91
92    field_nodes = {
93        field_node.name.value: field_node for field_node in node.fields
94    }
95
96    input_fields = input_object_type.input_fields
97
98    results = await asyncio.gather(
99        *[
100            input_field_value_coercer(
101                input_field,
102                parent_node,
103                field_nodes.get(input_field_name, UNDEFINED_VALUE),
104                ctx,
105                variables,
106                path=Path(path, input_field_name),
107            )
108            for input_field_name, input_field in input_fields.items()
109        ]
110    )
111
112    errors = []
113    coerced_values = {}
114    for input_field_name, input_field_result in zip(input_fields, results):
115        if input_field_result is SKIP_FIELD:
116            continue
117
118        if is_invalid_value(input_field_result):
119            return CoercionResult(value=UNDEFINED_VALUE)
120
121        input_field_value, input_field_errors = input_field_result
122        if is_invalid_value(input_field_value):
123            return CoercionResult(value=UNDEFINED_VALUE)
124        if input_field_errors:
125            errors.extend(input_field_errors)
126        elif not errors:
127            coerced_values[input_field_name] = input_field_value
128
129    return CoercionResult(value=coerced_values, errors=errors)
130