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