1from functools import partial 2from typing import Any, Callable, Dict, List, Optional, Union 3 4from tartiflette.coercers.outputs.compute import get_output_coercer 5from tartiflette.resolver.default import default_field_resolver 6from tartiflette.resolver.factory import resolve_field 7from tartiflette.types.helpers.get_directive_instances import ( 8 compute_directive_nodes, 9) 10from tartiflette.types.helpers.type import get_graphql_type 11from tartiflette.utils.directives import wraps_with_directives 12 13__all__ = ("GraphQLField",) 14 15 16class GraphQLField: 17 """ 18 Definition of a GraphQL field. 19 """ 20 21 # pylint: disable=too-many-instance-attributes 22 23 def __init__( 24 self, 25 name: str, 26 gql_type: Union["GraphQLList", "GraphQLNonNull", str], 27 arguments: Optional[Dict[str, "GraphQLArgument"]] = None, 28 resolver: Optional[Callable] = None, 29 description: Optional[str] = None, 30 directives: Optional[List["DirectiveNode"]] = None, 31 ) -> None: 32 """ 33 :param name: name of the field 34 :param gql_type: GraphQL type of the field 35 :param arguments: map of arguments linked to the field 36 :param resolver: callable in charge of resolving the field 37 :param description: description of the field 38 :param directives: list of directives linked to the field 39 :type name: str 40 :type gql_type: Union[GraphQLList, GraphQLNonNull, str] 41 :type arguments: Optional[Dict[str, GraphQLArgument]] 42 :type resolver: Optional[Callable] 43 :type description: Optional[str] 44 :type directives: Optional[List[DirectiveNode]] 45 """ 46 self.name = name 47 self.gql_type = gql_type 48 self.arguments = arguments or {} 49 self.description = description 50 self.graphql_type: Optional["GraphQLType"] = None 51 52 # Directives 53 self.directives = directives 54 self.on_post_bake: Optional[Callable] = None 55 self.introspection_directives: Optional[Callable] = None 56 57 # Resolvers 58 self.raw_resolver = resolver 59 self.resolver: Optional[Callable] = None 60 self.subscribe: Optional[Callable] = None 61 62 # Arguments coercer 63 self.arguments_coercer: Optional[Callable] = None 64 self.query_arguments_coercer: Optional[Callable] = None 65 self.subscription_arguments_coercer: Optional[Callable] = None 66 67 # Concurrently 68 self.list_concurrently: Optional[bool] = None 69 self.parent_concurrently: Optional[bool] = None 70 self.query_list_concurrently: Optional[bool] = None 71 self.query_parent_concurrently: Optional[bool] = None 72 self.subscription_list_concurrently: Optional[bool] = None 73 self.subscription_parent_concurrently: Optional[bool] = None 74 75 # Introspection attributes 76 self.isDeprecated: bool = False # pylint: disable=invalid-name 77 self.args: List["GraphQLArgument"] = [] 78 79 def __eq__(self, other: Any) -> bool: 80 """ 81 Returns True if `other` instance is identical to `self`. 82 :param other: object instance to compare to `self` 83 :type other: Any 84 :return: whether or not `other` is identical to `self` 85 :rtype: bool 86 """ 87 return self is other or ( 88 isinstance(other, GraphQLField) 89 and self.name == other.name 90 and self.gql_type == other.gql_type 91 and self.arguments == other.arguments 92 and self.description == other.description 93 and self.resolver == other.resolver 94 and self.directives == other.directives 95 ) 96 97 def __repr__(self) -> str: 98 """ 99 Returns the representation of a GraphQLField instance. 100 :return: the representation of a GraphQLField instance 101 :rtype: str 102 """ 103 return ( 104 "GraphQLField(name={!r}, gql_type={!r}, arguments={!r}, " 105 "resolver={!r}, description={!r}, directives={!r})".format( 106 self.name, 107 self.gql_type, 108 self.arguments, 109 self.resolver, 110 self.description, 111 self.directives, 112 ) 113 ) 114 115 def __str__(self) -> str: 116 """ 117 Returns a human-readable representation of the field. 118 :return: a human-readable representation of the field 119 :rtype: str 120 """ 121 return self.name 122 123 # Introspection attribute 124 @property 125 def kind(self) -> str: 126 """ 127 Returns the kind of the field which is used by the introspection query. 128 :return: the kind of the field 129 :rtype: str 130 """ 131 try: 132 return self.gql_type.kind 133 except AttributeError: 134 pass 135 return "FIELD" 136 137 # Introspection attribute 138 @property 139 def type(self) -> Union[str, "GraphQLType"]: 140 """ 141 Returns the GraphQL type of the field which is used by the 142 introspection query. 143 :return: the GraphQL type of the field 144 :rtype: Union[str, GraphQLType] 145 """ 146 return self.graphql_type 147 148 def bake( 149 self, 150 schema: "GraphQLSchema", 151 custom_default_resolver: Optional[Callable], 152 ) -> None: 153 """ 154 Bakes the GraphQLField and computes all the necessary stuff for 155 execution. 156 :param schema: the GraphQLSchema instance linked to the engine 157 :param custom_default_resolver: callable that will replace the builtin 158 default_resolver 159 :type schema: GraphQLSchema 160 :type custom_default_resolver: Optional[Callable] 161 """ 162 self.graphql_type = get_graphql_type(schema, self.gql_type) 163 164 if self.subscription_arguments_coercer is not None: 165 self.arguments_coercer = self.subscription_arguments_coercer 166 elif self.query_arguments_coercer is not None: 167 self.arguments_coercer = self.query_arguments_coercer 168 else: 169 self.arguments_coercer = schema.default_arguments_coercer 170 171 if self.subscription_list_concurrently is not None: 172 self.list_concurrently = self.subscription_list_concurrently 173 elif self.query_list_concurrently is not None: 174 self.list_concurrently = self.query_list_concurrently 175 else: 176 self.list_concurrently = schema.coerce_list_concurrently 177 178 if self.subscription_parent_concurrently is not None: 179 self.parent_concurrently = self.subscription_parent_concurrently 180 elif self.query_parent_concurrently is not None: 181 self.parent_concurrently = self.query_parent_concurrently 182 else: 183 self.parent_concurrently = schema.coerce_parent_concurrently 184 185 # Directives 186 directives_definition = compute_directive_nodes( 187 schema, self.directives 188 ) 189 self.on_post_bake = partial( 190 wraps_with_directives( 191 directives_definition=directives_definition, 192 directive_hook="on_post_bake", 193 with_default=True, 194 ), 195 self, 196 ) 197 self.introspection_directives = wraps_with_directives( 198 directives_definition=directives_definition, 199 directive_hook="on_introspection", 200 ) 201 202 # Resolvers 203 self.resolver = partial( 204 resolve_field, 205 field_definition=self, 206 resolver=wraps_with_directives( 207 directives_definition=directives_definition, 208 directive_hook="on_field_execution", 209 func=( 210 self.raw_resolver 211 or custom_default_resolver 212 or default_field_resolver 213 ), 214 is_resolver=True, 215 with_default=True, 216 ), 217 output_coercer=get_output_coercer( 218 self.graphql_type, self.list_concurrently 219 ), 220 ) 221 222 for argument in self.arguments.values(): 223 argument.bake(schema) 224 self.args.append(argument) 225