Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
:param parent_node: the root parent AST node
:param node: the AST node to treat
:param value: the raw value to compute
:param ctx: context passed to the query execution
:param input_field: the input field to compute
:param path: the path traveled until this coercer
:type parent_node: Union[VariableDefinitionNode, InputValueDefinitionNode]
:type node: Node
:type value: Any
:type ctx: Optional[Any]
:type input_field: GraphQLInputField
:type path: Path
:return: the coercion result
:rtype: Union[CoercionResult, UNDEFINED_VALUE]
"""
if is_invalid_value(value):
if input_field.default_value is not None:
return await input_field.literal_coercer(
parent_node, input_field.default_value, ctx
)
if input_field.graphql_type.is_non_null_type:
return CoercionResult(
errors=[
coercion_error(
f"Field < {path} > of required type "
f"< {input_field.gql_type} > was not provided",
node,
)
]
)
return UNDEFINED_VALUE
coercion_result = await coercer(
parent_node,
node,
ctx,
variables=variables,
path=path,
is_non_null_type=is_non_null_type,
)
if not directives or (
isinstance(node, VariableNode) and not is_input_field
):
return coercion_result
value, errors = coercion_result
if is_invalid_value(value) or errors:
return coercion_result
try:
return CoercionResult(
value=await directives(
parent_node, value, ctx, context_coercer=ctx
)
)
except Exception as raw_exception: # pylint: disable=broad-except
return CoercionResult(
errors=[
graphql_error_from_nodes(
str(raw_exception),
node,
original_error=(
raw_exception
# pylint: disable=too-many-locals
var_name = executable_variable_definition.name
var_type = executable_variable_definition.graphql_type
has_value = var_name in raw_variable_values
value = raw_variable_values.get(var_name, UNDEFINED_VALUE)
default_value = executable_variable_definition.default_value
variable_definition_node = executable_variable_definition.definition
if not has_value and not is_invalid_value(default_value):
# TODO: we should be able to remove this once the `ValuesOfCorrectType`
# rule of `validate_document` will be implemented
coercion_result = await literal_coercer(
variable_definition_node, default_value, ctx
)
value, _ = coercion_result
if is_invalid_value(value):
return CoercionResult(
errors=[
graphql_error_from_nodes(
f"Variable < ${var_name} > got invalid default value "
f"< {default_value} >.",
nodes=default_value,
)
]
)
return coercion_result
if (not has_value or value is None) and var_type.is_non_null_type:
return CoercionResult(
errors=[
graphql_error_from_nodes(
(
:param input_field: the input field to compute
:param parent_node: the root parent AST node
:param value_node: the value node to compute
:param ctx: context passed to the query execution
:param variables: the variables provided in the GraphQL request
:param path: the path traveled until this coercer
:type input_field: GraphQLInputField
:type parent_node: Union[VariableDefinitionNode, InputValueDefinitionNode]
:type value_node: Union[ValueNode, VariableNode, UNDEFINED_VALUE]
:type ctx: Optional[Any]
:type variables: Optional[Dict[str, Any]]
:type path: Optional[Path]
:return: the computed value
:rtype: Union[CoercionResult, UNDEFINED_VALUE, SKIP_FIELD]
"""
if is_invalid_value(value_node) or is_missing_variable(
value_node.value, variables
):
if input_field.default_value is not None:
input_field_node = input_field.default_value
elif input_field.graphql_type.is_non_null_type:
return UNDEFINED_VALUE
else:
return SKIP_FIELD
else:
input_field_node = value_node.value
return await input_field.literal_coercer(
parent_node, input_field_node, ctx, variables=variables, path=path
)
input_field_value_coercer(
parent_node,
node,
value.get(input_field_name, UNDEFINED_VALUE),
ctx,
input_field,
path=Path(path, input_field_name),
)
for input_field_name, input_field in input_fields.items()
]
)
errors = []
coerced_values = {}
for input_field_name, input_field_result in zip(input_fields, results):
if is_invalid_value(input_field_result):
continue
input_field_value, input_field_errors = input_field_result
if input_field_errors:
errors.extend(input_field_errors)
elif not errors:
coerced_values[input_field_name] = input_field_value
for input_field_name in value:
if input_field_name not in input_fields:
errors.append(
coercion_error(
f"Field < {input_field_name} > is not defined by type "
f"< {input_object_type.name} >",
node,
path,
:param scalar_type: the GraphQLScalarType instance of the scalar
:param variables: the variables provided in the GraphQL request
:param path: the path traveled until this coercer
:type parent_node: Union[VariableDefinitionNode, InputValueDefinitionNode]
:type node: Union[ValueNode, VariableNode]
:type ctx: Optional[Any]
:type scalar_type: GraphQLScalarType
:type variables: Optional[Dict[str, Any]]
:type path: Optional[Path]
:return: the computed value
:rtype: CoercionResult
"""
# pylint: disable=unused-argument
try:
value = scalar_type.parse_literal(node)
if not is_invalid_value(value):
return CoercionResult(value=value)
except Exception: # pylint: disable=broad-except
pass
return CoercionResult(value=UNDEFINED_VALUE)
:param ctx: context passed to the query execution
:param info: information related to the execution and the resolved field
:param context_coercer: context passed to the query execution to use on
argument coercion process
:type element: Any
:type ctx: Optional[Any]
:type info: ResolveInfo
:type context_coercer: Optional[Any]
:return: the coerced element
:rtype: Any
"""
if not isinstance(element, list):
computed_element = await execute_introspection_directive(
element, ctx, info, context_coercer=context_coercer
)
if not is_invalid_value(computed_element):
return computed_element
return None
results = await asyncio.gather(
*[
execute_introspection_directive(
item, ctx, info, context_coercer=context_coercer
)
for item in element
]
)
return [result for result in results if not is_invalid_value(result)]
computed_element = await execute_introspection_directive(
element, ctx, info, context_coercer=context_coercer
)
if not is_invalid_value(computed_element):
return computed_element
return None
results = await asyncio.gather(
*[
execute_introspection_directive(
item, ctx, info, context_coercer=context_coercer
)
for item in element
]
)
return [result for result in results if not is_invalid_value(result)]
parent_node,
item_node,
ctx,
is_non_null_item_type,
inner_coercer,
variables,
path=Path(path, index),
)
for index, item_node in enumerate(node.values)
]
)
errors = []
coerced_values = []
for coerced_result in results:
if is_invalid_value(coerced_result):
return CoercionResult(value=UNDEFINED_VALUE)
coerced_value, coerced_errors = coerced_result
if is_invalid_value(coerced_value):
return CoercionResult(value=UNDEFINED_VALUE)
if coerced_errors:
errors.extend(coerced_errors)
elif not errors:
coerced_values.append(coerced_value)
return CoercionResult(value=coerced_values, errors=errors)
coerced_item_value, coerced_item_errors = await inner_coercer(
parent_node, node, ctx, variables=variables, path=path
)
if is_invalid_value(coerced_item_value):
:type is_non_null_type: bool
:return: the computed value
:rtype: CoercionResult
"""
if not node:
return CoercionResult(value=UNDEFINED_VALUE)
if isinstance(node, NullValueNode):
return CoercionResult(value=None)
if isinstance(node, VariableNode):
if not variables:
return CoercionResult(value=UNDEFINED_VALUE)
value = variables.get(node.name.value, UNDEFINED_VALUE)
if is_invalid_value(value) or (value is None and is_non_null_type):
return CoercionResult(value=UNDEFINED_VALUE)
return CoercionResult(value=value)
return await coercer(
parent_node, node, ctx, variables=variables, **kwargs
)