1from functools import partial 2from typing import Any, Callable, Dict, List, Optional 3 4from tartiflette.coercers.arguments import coerce_arguments 5from tartiflette.utils.callables import ( 6 is_valid_async_generator, 7 is_valid_coroutine, 8) 9 10__all__ = ("compute_directive_nodes",) 11 12 13def get_callables(implementation: Any) -> Dict[str, Callable]: 14 """ 15 Computes a dictionary of all attribute name that starts with `on_` and are 16 linked to a callable. 17 :param implementation: the implementation to parse 18 :type implementation: Any 19 :return: a dictionary of attribute name and callable 20 :rtype: Dict[str, Callable] 21 """ 22 return { 23 key: getattr(implementation, key) 24 for key in dir(implementation) 25 if key.startswith("on_") 26 and ( 27 is_valid_coroutine(getattr(implementation, key)) 28 or is_valid_async_generator(getattr(implementation, key)) 29 ) 30 } 31 32 33def transform_directive( 34 directive: "GraphQLDirective", arguments_coercer: Callable 35) -> Dict[str, Any]: 36 """ 37 Transforms a directive definition into a dictionary of available callables 38 hooks and with default arguments. 39 :param directive: the directive definition to transform 40 :param arguments_coercer: callable to use to coerce directive arguments 41 :type directive: GraphQLDirective 42 :type arguments_coercer: Callable 43 :return: the transformed directive definition 44 :rtype: Dict[str, Any] 45 """ 46 return { 47 "callables": get_callables(directive.implementation), 48 "arguments_coercer": arguments_coercer, 49 } 50 51 52def compute_directive_nodes( 53 schema: "GraphQLSchema", 54 directive_nodes: List["DirectiveNode"], 55 variable_values: Optional[Dict[str, Any]] = None, 56) -> List[Dict[str, Any]]: 57 """ 58 Computes a list of AST directive node into a list of pre-computed 59 directives. 60 :param schema: the GraphQLSchema instance linked to the engine 61 :param directive_nodes: list of AST directive node to compute 62 :param variable_values: the variables provided in the GraphQL request 63 :type schema: GraphQLSchema 64 :type directive_nodes: List[DirectiveNode] 65 :type variable_values: Optional[Dict[str, Any]] 66 :return: list of pre-computed directives 67 :rtype: List[Dict[str, Any]] 68 """ 69 if not directive_nodes: 70 return [] 71 72 computed_directives = [] 73 for directive_node in directive_nodes: 74 directive_definition = schema.find_directive(directive_node.name.value) 75 computed_directives.append( 76 transform_directive( 77 directive_definition, 78 arguments_coercer=partial( 79 coerce_arguments, 80 argument_definitions=directive_definition.arguments, 81 node=directive_node, 82 variable_values=variable_values or {}, 83 coercer=directive_definition.arguments_coercer, 84 ), 85 ) 86 ) 87 return computed_directives 88