1import asyncio 2 3from functools import partial 4from typing import Any, AsyncGenerator, Callable, Dict, List, Optional, Union 5 6from tartiflette.constants import UNDEFINED_VALUE 7from tartiflette.utils.values import is_invalid_value 8 9__all__ = ("introspection_directives_executor", "wraps_with_directives") 10 11 12async def execute_introspection_directive( 13 element: Any, 14 ctx: Optional[Any], 15 info: "ResolveInfo", 16 context_coercer: Optional[Any] = None, 17) -> Any: 18 """ 19 Applies introspection directives on the element. 20 :param element: element to treat 21 :param ctx: context passed to the query execution 22 :param info: information related to the execution and the resolved field 23 :param context_coercer: context passed to the query execution to use on 24 argument coercion process 25 :type element: Any 26 :type ctx: Optional[Any] 27 :type info: ResolveInfo 28 :type context_coercer: Optional[Any] 29 :return: the coerced element 30 :rtype: Any 31 """ 32 try: 33 if element.introspection_directives: 34 result = await element.introspection_directives( 35 element, ctx, info, context_coercer=context_coercer 36 ) 37 if result: 38 return result 39 return UNDEFINED_VALUE 40 except (AttributeError, TypeError): 41 pass 42 return element 43 44 45async def introspection_directives_executor( 46 element: Any, 47 ctx: Optional[Any], 48 info: "ResolveInfo", 49 context_coercer: Optional[Any] = None, 50) -> Any: 51 """ 52 Applies introspection directives on the element or a list of elements. 53 :param element: element to treat 54 :param ctx: context passed to the query execution 55 :param info: information related to the execution and the resolved field 56 :param context_coercer: context passed to the query execution to use on 57 argument coercion process 58 :type element: Any 59 :type ctx: Optional[Any] 60 :type info: ResolveInfo 61 :type context_coercer: Optional[Any] 62 :return: the coerced element 63 :rtype: Any 64 """ 65 if not isinstance(element, list): 66 computed_element = await execute_introspection_directive( 67 element, ctx, info, context_coercer=context_coercer 68 ) 69 if not is_invalid_value(computed_element): 70 return computed_element 71 return None 72 73 results = await asyncio.gather( 74 *[ 75 execute_introspection_directive( 76 item, ctx, info, context_coercer=context_coercer 77 ) 78 for item in element 79 ] 80 ) 81 return [result for result in results if not is_invalid_value(result)] 82 83 84async def default_argument_execution_directive( 85 parent_node: Union["FieldNode", "DirectiveNode"], 86 argument_definition_node: "InputValueDefinitionNode", 87 argument_node: Optional["ArgumentNode"], 88 value: Any, 89 *args, 90 **kwargs, 91) -> Any: 92 """ 93 Default callable to use to wrap with directives on `on_argument_execution` 94 hook name. 95 :param parent_node: the parent AST node related to the executed argument 96 :param argument_definition_node: the input value definition AST node 97 :param argument_node: the AST argument node executed 98 :param value: the coerced value of the argument 99 :type parent_node: Union[FieldNode, DirectiveNode] 100 :type argument_definition_node: InputValueDefinitionNode 101 :type argument_node: ArgumentNode 102 :type value: Any 103 :return: the coerced value of the argument 104 :rtype: Any 105 """ 106 # pylint: disable=unused-argument 107 return value 108 109 110async def default_post_input_coercion_directive( 111 parent_node: Union["VariableDefinitionNode", "InputValueDefinitionNode"], 112 value: Any, 113 *args, 114 **kwargs, 115) -> Any: 116 """ 117 Default callable to use to wrap with directives on `on_post_input_coercion` 118 hook name. 119 :param parent_node: the root parent AST node 120 :param value: the coerced value of the argument 121 :type parent_node: Union[VariableDefinitionNode, InputValueDefinitionNode] 122 :type value: Any 123 :return: the coerced value of the argument 124 :rtype: Any 125 """ 126 # pylint: disable=unused-argument 127 return value 128 129 130async def default_directive_callable(value: Any, *args, **kwargs) -> Any: 131 """ 132 Default callable to use to wrap with directives when the hook doesn't 133 implements a specific callable. 134 :param value: the coerced value 135 :type value: Any 136 :return: the coerced value 137 :rtype: Any 138 """ 139 # pylint: disable=unused-argument 140 return value 141 142 143_HOOK_CALLABLES_MAP = { 144 "on_argument_execution": default_argument_execution_directive, 145 "on_post_input_coercion": default_post_input_coercion_directive, 146} 147 148 149async def directive_executor( 150 directive_func: Callable, 151 directive_arguments_coercer: Callable, 152 wrapped_func: Callable, 153 *args, 154 context_coercer: Optional[Any] = None, 155 **kwargs, 156) -> Any: 157 """ 158 Wraps the execution of directives in order to handle properly the fact that 159 directive arguments can be a dictionary or a callable. 160 :param directive_func: callable representing the directive implementation 161 :param directive_arguments_coercer: callable to use to coerce directive 162 arguments 163 :param context_coercer: context passed to the query execution to use on 164 argument coercion process 165 :param wrapped_func: the inner callable to call after the directive 166 :type directive_func: Callable 167 :type directive_arguments_coercer: Callable 168 :type wrapped_func: Callable 169 :type context_coercer: Optional[Any] 170 :return: the computed value 171 :rtype: Any 172 """ 173 return await directive_func( 174 await directive_arguments_coercer(ctx=context_coercer), 175 partial(wrapped_func, context_coercer=context_coercer), 176 *args, 177 **kwargs, 178 ) 179 180 181async def directive_generator( 182 directive_func: AsyncGenerator, 183 directive_arguments_coercer: Callable, 184 wrapped_func: Callable, 185 *args, 186 context_coercer: Optional[Any] = None, 187 **kwargs, 188): 189 async for payload in directive_func( 190 await directive_arguments_coercer(ctx=context_coercer), 191 partial(wrapped_func, context_coercer=context_coercer), 192 *args, 193 **kwargs, 194 ): 195 yield payload 196 197 198async def resolver_executor(resolver: Callable, *args, **kwargs) -> Any: 199 """ 200 Wraos the execution of the raw resolver in order to pop the 201 `context_coercer` keyword arguments to avoid exception. 202 :param resolver: callable to wrap 203 :type resolver: Callable 204 :return: resolved value 205 :rtype: Any 206 """ 207 kwargs.pop("context_coercer", None) 208 return await resolver(*args, **kwargs) 209 210 211async def subscription_generator(generator: AsyncGenerator, *args, **kwargs): 212 kwargs.pop("context_coercer", None) 213 async for payload in generator(*args, **kwargs): 214 yield payload 215 216 217def wraps_with_directives( 218 directives_definition: List[Dict[str, Any]], 219 directive_hook: str, 220 func: Optional[Callable] = None, 221 is_resolver: bool = False, 222 with_default: bool = False, 223 is_async_generator: bool = False, 224) -> Optional[Callable]: 225 """ 226 Wraps a callable with directives. 227 :param directives_definition: directives to wrap with 228 :param directive_hook: name of the hook to wrap with 229 :param func: callable to wrap 230 :param is_resolver: determines whether or not the wrapped func is a 231 resolver 232 :param with_default: determines whether or not if there is no directives 233 :param is_async_generator: determines whether or not the wrapped func is a generator 234 definition we should return or not a callable 235 :type directives_definition: List[Dict[str, Any]] 236 :type directive_hook: str 237 :type func: Optional[Callable] 238 :type is_resolver: bool 239 :type with_default: bool 240 :type is_async_generator: bool 241 :return: wrapped callable 242 :rtype: Optional[Callable] 243 """ 244 directive_wrapper = directive_executor 245 246 if func is None: 247 if not with_default and not directives_definition: 248 return None 249 250 func = _HOOK_CALLABLES_MAP.get( 251 directive_hook, default_directive_callable 252 ) 253 254 if is_resolver and not isinstance(func, partial): 255 func = partial(resolver_executor, func) 256 257 if is_async_generator and not isinstance(func, partial): 258 func = partial(subscription_generator, func) 259 directive_wrapper = directive_generator 260 261 for directive in reversed(directives_definition): 262 if directive_hook in directive["callables"]: 263 func = partial( 264 directive_wrapper, 265 directive["callables"][directive_hook], 266 directive["arguments_coercer"], 267 func, 268 ) 269 return func 270