We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Slow performance in async execution because of DirectivesExtension
The use of DirectivesExtension in async graphql execution wraps every field of the result graph in an async function. This slows down performance. 
We have one resolver returning a huge list of objects where every field of the model is already filled with a native type (str, int ..). So there is no need for an asynchronous resolve of these model fields. 
But the DirectivesExtension calls `await_maybe` for each field of each model in the result list. Now there are thousands of async function calls which should be awaited. 
We don't use custom directives so I replaced the Schema.execute function (from strawberry.schema.schema.py). In this function I only register the DirectivesExtension if there are custom directives. For us this solves the problem, as long as we don't use custom directives. But I think it would be a modification that could also be merged in the strawberry sourcecode. 

Here is a demo script with the used patch, showing the dramatic performance improvement for this case. 

``` 
import asyncio
import time
import typing
from typing import Any, Dict, Optional

import strawberry
from strawberry import Schema
from strawberry.extensions.directives import DirectivesExtension
from strawberry.schema.execute import execute
from strawberry.types import ExecutionContext, ExecutionResult


@strawberry.type
class Book:
    author: str
    title: str


async def get_books():
    res = []
    await asyncio.sleep(0.1)
    for i in range(200000):
        res.append(
            Book(
                author="Unknown",
                title=f"Book nr {i}",
            )
        )
    return res


@strawberry.type
class DemoQuery:
    books: typing.List[Book] = strawberry.field(resolver=get_books)


async def test():
    start_time = time.time()
    schema = strawberry.Schema(query=DemoQuery)
    query = "{ books { title author } }"
    result = await schema.execute(
        query,
    )
    if result.errors:
        print(result.errors)

    print("complete", time.time() - start_time)


async def schema_execute(
    self,
    query: str,
    variable_values: Optional[Dict[str, Any]] = None,
    context_value: Optional[Any] = None,
    root_value: Optional[Any] = None,
    operation_name: Optional[str] = None,
) -> ExecutionResult:
    # Create execution context
    execution_context = ExecutionContext(
        query=query,
        schema=self,
        context=context_value,
        root_value=root_value,
        variables=variable_values,
        provided_operation_name=operation_name,
    )

    # BEGIN - CHANGED
    extensions = list(self.extensions)
    if len(self.directives) > 0:
        extensions += [DirectivesExtension]

    result = await execute(
        self._schema,  # pylint: disable=protected-access
        query,
        extensions=extensions,
        execution_context_class=self.execution_context_class,
        execution_context=execution_context,
    )
    # END - CHANGED

    if result.errors:
        self.process_errors(result.errors, execution_context=execution_context)

    return result


def apply_directives_patch():
    Schema.execute = schema_execute


def run():
    asyncio.run(test())


def run_with_patch():
    apply_directives_patch()
    asyncio.run(test())


if __name__ == "__main__":
    print("run")
    run()
    print("run with patch")
    run_with_patch()

``` 

On my Machine I get the following output:

```
run
complete 24.10178303718567
run with patch
complete 4.264432191848755
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `strawberry/schema/schema.py`
Content:
```
1 from functools import lru_cache
2 from typing import Any, Dict, Iterable, Optional, Sequence, Type, Union
3 
4 from graphql import (
5     ExecutionContext as GraphQLExecutionContext,
6     GraphQLNamedType,
7     GraphQLNonNull,
8     GraphQLSchema,
9     get_introspection_query,
10     parse,
11     validate_schema,
12 )
13 from graphql.subscription import subscribe
14 from graphql.type.directives import specified_directives
15 
16 from strawberry.custom_scalar import ScalarDefinition, ScalarWrapper
17 from strawberry.directive import StrawberryDirective
18 from strawberry.enum import EnumDefinition
19 from strawberry.extensions import Extension
20 from strawberry.extensions.directives import (
21     DirectivesExtension,
22     DirectivesExtensionSync,
23 )
24 from strawberry.field import StrawberryField
25 from strawberry.schema.schema_converter import GraphQLCoreConverter
26 from strawberry.schema.types.scalar import DEFAULT_SCALAR_REGISTRY
27 from strawberry.types import ExecutionContext, ExecutionResult
28 from strawberry.types.graphql import OperationType
29 from strawberry.types.types import TypeDefinition
30 from strawberry.union import StrawberryUnion
31 
32 from ..printer import print_schema
33 from .base import BaseSchema
34 from .config import StrawberryConfig
35 from .execute import execute, execute_sync
36 
37 
38 DEFAULT_ALLOWED_OPERATION_TYPES = {
39     OperationType.QUERY,
40     OperationType.MUTATION,
41     OperationType.SUBSCRIPTION,
42 }
43 
44 
45 class Schema(BaseSchema):
46     def __init__(
47         self,
48         # TODO: can we make sure we only allow to pass something that has been decorated?
49         query: Type,
50         mutation: Optional[Type] = None,
51         subscription: Optional[Type] = None,
52         directives: Sequence[StrawberryDirective] = (),
53         types=(),
54         extensions: Sequence[Union[Type[Extension], Extension]] = (),
55         execution_context_class: Optional[Type[GraphQLExecutionContext]] = None,
56         config: Optional[StrawberryConfig] = None,
57         scalar_overrides: Optional[
58             Dict[object, Union[ScalarWrapper, ScalarDefinition]]
59         ] = None,
60     ):
61         self.extensions = extensions
62         self.execution_context_class = execution_context_class
63         self.config = config or StrawberryConfig()
64 
65         scalar_registry: Dict[object, Union[ScalarWrapper, ScalarDefinition]] = {
66             **DEFAULT_SCALAR_REGISTRY
67         }
68         if scalar_overrides:
69             scalar_registry.update(scalar_overrides)
70 
71         self.schema_converter = GraphQLCoreConverter(self.config, scalar_registry)
72         self.directives = directives
73 
74         query_type = self.schema_converter.from_object(query._type_definition)
75         mutation_type = (
76             self.schema_converter.from_object(mutation._type_definition)
77             if mutation
78             else None
79         )
80         subscription_type = (
81             self.schema_converter.from_object(subscription._type_definition)
82             if subscription
83             else None
84         )
85 
86         graphql_directives = tuple(
87             self.schema_converter.from_directive(directive) for directive in directives
88         )
89 
90         graphql_types = []
91         for type_ in types:
92             graphql_type = self.schema_converter.from_maybe_optional(type_)
93             if isinstance(graphql_type, GraphQLNonNull):
94                 graphql_type = graphql_type.of_type
95             if not isinstance(graphql_type, GraphQLNamedType):
96                 raise TypeError(f"{graphql_type} is not a named GraphQL Type")
97             graphql_types.append(graphql_type)
98 
99         self._schema = GraphQLSchema(
100             query=query_type,
101             mutation=mutation_type,
102             subscription=subscription_type if subscription else None,
103             directives=specified_directives + graphql_directives,
104             types=graphql_types,
105             extensions={
106                 GraphQLCoreConverter.DEFINITION_BACKREF: self,
107             },
108         )
109 
110         # attach our schema to the GraphQL schema instance
111         self._schema._strawberry_schema = self  # type: ignore
112 
113         # Validate schema early because we want developers to know about
114         # possible issues as soon as possible
115         errors = validate_schema(self._schema)
116         if errors:
117             formatted_errors = "\n\n".join(f"❌ {error.message}" for error in errors)
118             raise ValueError(f"Invalid Schema. Errors:\n\n{formatted_errors}")
119 
120         self.query = self.schema_converter.type_map[query_type.name]
121 
122     @lru_cache()
123     def get_type_by_name(  # type: ignore  # lru_cache makes mypy complain
124         self, name: str
125     ) -> Optional[
126         Union[TypeDefinition, ScalarDefinition, EnumDefinition, StrawberryUnion]
127     ]:
128         # TODO: respect auto_camel_case
129         if name in self.schema_converter.type_map:
130             return self.schema_converter.type_map[name].definition
131 
132         return None
133 
134     def get_field_for_type(
135         self, field_name: str, type_name: str
136     ) -> Optional[StrawberryField]:
137         type_ = self.get_type_by_name(type_name)
138 
139         if not type_:
140             return None  # pragma: no cover
141 
142         assert isinstance(type_, TypeDefinition)
143 
144         return next(
145             (
146                 field
147                 for field in type_.fields
148                 if self.config.name_converter.get_graphql_name(field) == field_name
149             ),
150             None,
151         )
152 
153     @lru_cache()
154     def get_directive_by_name(self, graphql_name: str) -> Optional[StrawberryDirective]:
155         return next(
156             (
157                 directive
158                 for directive in self.directives
159                 if self.config.name_converter.from_directive(directive) == graphql_name
160             ),
161             None,
162         )
163 
164     async def execute(
165         self,
166         query: str,
167         variable_values: Optional[Dict[str, Any]] = None,
168         context_value: Optional[Any] = None,
169         root_value: Optional[Any] = None,
170         operation_name: Optional[str] = None,
171         allowed_operation_types: Optional[Iterable[OperationType]] = None,
172     ) -> ExecutionResult:
173         if allowed_operation_types is None:
174             allowed_operation_types = DEFAULT_ALLOWED_OPERATION_TYPES
175 
176         # Create execution context
177         execution_context = ExecutionContext(
178             query=query,
179             schema=self,
180             context=context_value,
181             root_value=root_value,
182             variables=variable_values,
183             provided_operation_name=operation_name,
184         )
185 
186         result = await execute(
187             self._schema,
188             query,
189             extensions=list(self.extensions) + [DirectivesExtension],
190             execution_context_class=self.execution_context_class,
191             execution_context=execution_context,
192             allowed_operation_types=allowed_operation_types,
193         )
194 
195         if result.errors:
196             self.process_errors(result.errors, execution_context=execution_context)
197 
198         return result
199 
200     def execute_sync(
201         self,
202         query: str,
203         variable_values: Optional[Dict[str, Any]] = None,
204         context_value: Optional[Any] = None,
205         root_value: Optional[Any] = None,
206         operation_name: Optional[str] = None,
207         allowed_operation_types: Optional[Iterable[OperationType]] = None,
208     ) -> ExecutionResult:
209         if allowed_operation_types is None:
210             allowed_operation_types = DEFAULT_ALLOWED_OPERATION_TYPES
211 
212         execution_context = ExecutionContext(
213             query=query,
214             schema=self,
215             context=context_value,
216             root_value=root_value,
217             variables=variable_values,
218             provided_operation_name=operation_name,
219         )
220 
221         result = execute_sync(
222             self._schema,
223             query,
224             extensions=list(self.extensions) + [DirectivesExtensionSync],
225             execution_context_class=self.execution_context_class,
226             execution_context=execution_context,
227             allowed_operation_types=allowed_operation_types,
228         )
229 
230         if result.errors:
231             self.process_errors(result.errors, execution_context=execution_context)
232 
233         return result
234 
235     async def subscribe(
236         self,
237         query: str,
238         variable_values: Optional[Dict[str, Any]] = None,
239         context_value: Optional[Any] = None,
240         root_value: Optional[Any] = None,
241         operation_name: Optional[str] = None,
242     ):
243         return await subscribe(
244             self._schema,
245             parse(query),
246             root_value=root_value,
247             context_value=context_value,
248             variable_values=variable_values,
249             operation_name=operation_name,
250         )
251 
252     def as_str(self) -> str:
253         return print_schema(self)
254 
255     __str__ = as_str
256 
257     def introspect(self) -> Dict[str, Any]:
258         """Return the introspection query result for the current schema
259 
260         Raises:
261             ValueError: If the introspection query fails due to an invalid schema
262         """
263         introspection = self.execute_sync(get_introspection_query())
264         if introspection.errors or not introspection.data:
265             raise ValueError(f"Invalid Schema. Errors {introspection.errors!r}")
266 
267         return introspection.data
268
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/strawberry/schema/schema.py b/strawberry/schema/schema.py
--- a/strawberry/schema/schema.py
+++ b/strawberry/schema/schema.py
@@ -1,5 +1,5 @@
 from functools import lru_cache
-from typing import Any, Dict, Iterable, Optional, Sequence, Type, Union
+from typing import Any, Dict, Iterable, List, Optional, Sequence, Type, Union
 
 from graphql import (
     ExecutionContext as GraphQLExecutionContext,
@@ -119,6 +119,16 @@
 
         self.query = self.schema_converter.type_map[query_type.name]
 
+    def get_extensions(
+        self, sync: bool = False
+    ) -> List[Union[Type[Extension], Extension]]:
+        extensions = list(self.extensions)
+
+        if self.directives:
+            extensions.append(DirectivesExtensionSync if sync else DirectivesExtension)
+
+        return extensions
+
     @lru_cache()
     def get_type_by_name(  # type: ignore  # lru_cache makes mypy complain
         self, name: str
@@ -186,7 +196,7 @@
         result = await execute(
             self._schema,
             query,
-            extensions=list(self.extensions) + [DirectivesExtension],
+            extensions=self.get_extensions(),
             execution_context_class=self.execution_context_class,
             execution_context=execution_context,
             allowed_operation_types=allowed_operation_types,
@@ -221,7 +231,7 @@
         result = execute_sync(
             self._schema,
             query,
-            extensions=list(self.extensions) + [DirectivesExtensionSync],
+            extensions=self.get_extensions(sync=True),
             execution_context_class=self.execution_context_class,
             execution_context=execution_context,
             allowed_operation_types=allowed_operation_types,

{"golden_diff": "diff --git a/strawberry/schema/schema.py b/strawberry/schema/schema.py\n--- a/strawberry/schema/schema.py\n+++ b/strawberry/schema/schema.py\n@@ -1,5 +1,5 @@\n from functools import lru_cache\n-from typing import Any, Dict, Iterable, Optional, Sequence, Type, Union\n+from typing import Any, Dict, Iterable, List, Optional, Sequence, Type, Union\n \n from graphql import (\n     ExecutionContext as GraphQLExecutionContext,\n@@ -119,6 +119,16 @@\n \n         self.query = self.schema_converter.type_map[query_type.name]\n \n+    def get_extensions(\n+        self, sync: bool = False\n+    ) -> List[Union[Type[Extension], Extension]]:\n+        extensions = list(self.extensions)\n+\n+        if self.directives:\n+            extensions.append(DirectivesExtensionSync if sync else DirectivesExtension)\n+\n+        return extensions\n+\n     @lru_cache()\n     def get_type_by_name(  # type: ignore  # lru_cache makes mypy complain\n         self, name: str\n@@ -186,7 +196,7 @@\n         result = await execute(\n             self._schema,\n             query,\n-            extensions=list(self.extensions) + [DirectivesExtension],\n+            extensions=self.get_extensions(),\n             execution_context_class=self.execution_context_class,\n             execution_context=execution_context,\n             allowed_operation_types=allowed_operation_types,\n@@ -221,7 +231,7 @@\n         result = execute_sync(\n             self._schema,\n             query,\n-            extensions=list(self.extensions) + [DirectivesExtensionSync],\n+            extensions=self.get_extensions(sync=True),\n             execution_context_class=self.execution_context_class,\n             execution_context=execution_context,\n             allowed_operation_types=allowed_operation_types,\n", "issue": "Slow performance in async execution because of DirectivesExtension\nThe use of DirectivesExtension in async graphql execution wraps every field of the result graph in an async function. This slows down performance. \r\nWe have one resolver returning a huge list of objects where every field of the model is already filled with a native type (str, int ..). So there is no need for an asynchronous resolve of these model fields. \r\nBut the DirectivesExtension calls `await_maybe` for each field of each model in the result list. Now there are thousands of async function calls which should be awaited. \r\nWe don't use custom directives so I replaced the Schema.execute function (from strawberry.schema.schema.py). In this function I only register the DirectivesExtension if there are custom directives. For us this solves the problem, as long as we don't use custom directives. But I think it would be a modification that could also be merged in the strawberry sourcecode. \r\n\r\nHere is a demo script with the used patch, showing the dramatic performance improvement for this case. \r\n\r\n``` \r\nimport asyncio\r\nimport time\r\nimport typing\r\nfrom typing import Any, Dict, Optional\r\n\r\nimport strawberry\r\nfrom strawberry import Schema\r\nfrom strawberry.extensions.directives import DirectivesExtension\r\nfrom strawberry.schema.execute import execute\r\nfrom strawberry.types import ExecutionContext, ExecutionResult\r\n\r\n\r\n@strawberry.type\r\nclass Book:\r\n    author: str\r\n    title: str\r\n\r\n\r\nasync def get_books():\r\n    res = []\r\n    await asyncio.sleep(0.1)\r\n    for i in range(200000):\r\n        res.append(\r\n            Book(\r\n                author=\"Unknown\",\r\n                title=f\"Book nr {i}\",\r\n            )\r\n        )\r\n    return res\r\n\r\n\r\n@strawberry.type\r\nclass DemoQuery:\r\n    books: typing.List[Book] = strawberry.field(resolver=get_books)\r\n\r\n\r\nasync def test():\r\n    start_time = time.time()\r\n    schema = strawberry.Schema(query=DemoQuery)\r\n    query = \"{ books { title author } }\"\r\n    result = await schema.execute(\r\n        query,\r\n    )\r\n    if result.errors:\r\n        print(result.errors)\r\n\r\n    print(\"complete\", time.time() - start_time)\r\n\r\n\r\nasync def schema_execute(\r\n    self,\r\n    query: str,\r\n    variable_values: Optional[Dict[str, Any]] = None,\r\n    context_value: Optional[Any] = None,\r\n    root_value: Optional[Any] = None,\r\n    operation_name: Optional[str] = None,\r\n) -> ExecutionResult:\r\n    # Create execution context\r\n    execution_context = ExecutionContext(\r\n        query=query,\r\n        schema=self,\r\n        context=context_value,\r\n        root_value=root_value,\r\n        variables=variable_values,\r\n        provided_operation_name=operation_name,\r\n    )\r\n\r\n    # BEGIN - CHANGED\r\n    extensions = list(self.extensions)\r\n    if len(self.directives) > 0:\r\n        extensions += [DirectivesExtension]\r\n\r\n    result = await execute(\r\n        self._schema,  # pylint: disable=protected-access\r\n        query,\r\n        extensions=extensions,\r\n        execution_context_class=self.execution_context_class,\r\n        execution_context=execution_context,\r\n    )\r\n    # END - CHANGED\r\n\r\n    if result.errors:\r\n        self.process_errors(result.errors, execution_context=execution_context)\r\n\r\n    return result\r\n\r\n\r\ndef apply_directives_patch():\r\n    Schema.execute = schema_execute\r\n\r\n\r\ndef run():\r\n    asyncio.run(test())\r\n\r\n\r\ndef run_with_patch():\r\n    apply_directives_patch()\r\n    asyncio.run(test())\r\n\r\n\r\nif __name__ == \"__main__\":\r\n    print(\"run\")\r\n    run()\r\n    print(\"run with patch\")\r\n    run_with_patch()\r\n\r\n``` \r\n\r\nOn my Machine I get the following output:\r\n\r\n```\r\nrun\r\ncomplete 24.10178303718567\r\nrun with patch\r\ncomplete 4.264432191848755\r\n```\r\n\n", "before_files": [{"content": "from functools import lru_cache\nfrom typing import Any, Dict, Iterable, Optional, Sequence, Type, Union\n\nfrom graphql import (\n    ExecutionContext as GraphQLExecutionContext,\n    GraphQLNamedType,\n    GraphQLNonNull,\n    GraphQLSchema,\n    get_introspection_query,\n    parse,\n    validate_schema,\n)\nfrom graphql.subscription import subscribe\nfrom graphql.type.directives import specified_directives\n\nfrom strawberry.custom_scalar import ScalarDefinition, ScalarWrapper\nfrom strawberry.directive import StrawberryDirective\nfrom strawberry.enum import EnumDefinition\nfrom strawberry.extensions import Extension\nfrom strawberry.extensions.directives import (\n    DirectivesExtension,\n    DirectivesExtensionSync,\n)\nfrom strawberry.field import StrawberryField\nfrom strawberry.schema.schema_converter import GraphQLCoreConverter\nfrom strawberry.schema.types.scalar import DEFAULT_SCALAR_REGISTRY\nfrom strawberry.types import ExecutionContext, ExecutionResult\nfrom strawberry.types.graphql import OperationType\nfrom strawberry.types.types import TypeDefinition\nfrom strawberry.union import StrawberryUnion\n\nfrom ..printer import print_schema\nfrom .base import BaseSchema\nfrom .config import StrawberryConfig\nfrom .execute import execute, execute_sync\n\n\nDEFAULT_ALLOWED_OPERATION_TYPES = {\n    OperationType.QUERY,\n    OperationType.MUTATION,\n    OperationType.SUBSCRIPTION,\n}\n\n\nclass Schema(BaseSchema):\n    def __init__(\n        self,\n        # TODO: can we make sure we only allow to pass something that has been decorated?\n        query: Type,\n        mutation: Optional[Type] = None,\n        subscription: Optional[Type] = None,\n        directives: Sequence[StrawberryDirective] = (),\n        types=(),\n        extensions: Sequence[Union[Type[Extension], Extension]] = (),\n        execution_context_class: Optional[Type[GraphQLExecutionContext]] = None,\n        config: Optional[StrawberryConfig] = None,\n        scalar_overrides: Optional[\n            Dict[object, Union[ScalarWrapper, ScalarDefinition]]\n        ] = None,\n    ):\n        self.extensions = extensions\n        self.execution_context_class = execution_context_class\n        self.config = config or StrawberryConfig()\n\n        scalar_registry: Dict[object, Union[ScalarWrapper, ScalarDefinition]] = {\n            **DEFAULT_SCALAR_REGISTRY\n        }\n        if scalar_overrides:\n            scalar_registry.update(scalar_overrides)\n\n        self.schema_converter = GraphQLCoreConverter(self.config, scalar_registry)\n        self.directives = directives\n\n        query_type = self.schema_converter.from_object(query._type_definition)\n        mutation_type = (\n            self.schema_converter.from_object(mutation._type_definition)\n            if mutation\n            else None\n        )\n        subscription_type = (\n            self.schema_converter.from_object(subscription._type_definition)\n            if subscription\n            else None\n        )\n\n        graphql_directives = tuple(\n            self.schema_converter.from_directive(directive) for directive in directives\n        )\n\n        graphql_types = []\n        for type_ in types:\n            graphql_type = self.schema_converter.from_maybe_optional(type_)\n            if isinstance(graphql_type, GraphQLNonNull):\n                graphql_type = graphql_type.of_type\n            if not isinstance(graphql_type, GraphQLNamedType):\n                raise TypeError(f\"{graphql_type} is not a named GraphQL Type\")\n            graphql_types.append(graphql_type)\n\n        self._schema = GraphQLSchema(\n            query=query_type,\n            mutation=mutation_type,\n            subscription=subscription_type if subscription else None,\n            directives=specified_directives + graphql_directives,\n            types=graphql_types,\n            extensions={\n                GraphQLCoreConverter.DEFINITION_BACKREF: self,\n            },\n        )\n\n        # attach our schema to the GraphQL schema instance\n        self._schema._strawberry_schema = self  # type: ignore\n\n        # Validate schema early because we want developers to know about\n        # possible issues as soon as possible\n        errors = validate_schema(self._schema)\n        if errors:\n            formatted_errors = \"\\n\\n\".join(f\"\u274c {error.message}\" for error in errors)\n            raise ValueError(f\"Invalid Schema. Errors:\\n\\n{formatted_errors}\")\n\n        self.query = self.schema_converter.type_map[query_type.name]\n\n    @lru_cache()\n    def get_type_by_name(  # type: ignore  # lru_cache makes mypy complain\n        self, name: str\n    ) -> Optional[\n        Union[TypeDefinition, ScalarDefinition, EnumDefinition, StrawberryUnion]\n    ]:\n        # TODO: respect auto_camel_case\n        if name in self.schema_converter.type_map:\n            return self.schema_converter.type_map[name].definition\n\n        return None\n\n    def get_field_for_type(\n        self, field_name: str, type_name: str\n    ) -> Optional[StrawberryField]:\n        type_ = self.get_type_by_name(type_name)\n\n        if not type_:\n            return None  # pragma: no cover\n\n        assert isinstance(type_, TypeDefinition)\n\n        return next(\n            (\n                field\n                for field in type_.fields\n                if self.config.name_converter.get_graphql_name(field) == field_name\n            ),\n            None,\n        )\n\n    @lru_cache()\n    def get_directive_by_name(self, graphql_name: str) -> Optional[StrawberryDirective]:\n        return next(\n            (\n                directive\n                for directive in self.directives\n                if self.config.name_converter.from_directive(directive) == graphql_name\n            ),\n            None,\n        )\n\n    async def execute(\n        self,\n        query: str,\n        variable_values: Optional[Dict[str, Any]] = None,\n        context_value: Optional[Any] = None,\n        root_value: Optional[Any] = None,\n        operation_name: Optional[str] = None,\n        allowed_operation_types: Optional[Iterable[OperationType]] = None,\n    ) -> ExecutionResult:\n        if allowed_operation_types is None:\n            allowed_operation_types = DEFAULT_ALLOWED_OPERATION_TYPES\n\n        # Create execution context\n        execution_context = ExecutionContext(\n            query=query,\n            schema=self,\n            context=context_value,\n            root_value=root_value,\n            variables=variable_values,\n            provided_operation_name=operation_name,\n        )\n\n        result = await execute(\n            self._schema,\n            query,\n            extensions=list(self.extensions) + [DirectivesExtension],\n            execution_context_class=self.execution_context_class,\n            execution_context=execution_context,\n            allowed_operation_types=allowed_operation_types,\n        )\n\n        if result.errors:\n            self.process_errors(result.errors, execution_context=execution_context)\n\n        return result\n\n    def execute_sync(\n        self,\n        query: str,\n        variable_values: Optional[Dict[str, Any]] = None,\n        context_value: Optional[Any] = None,\n        root_value: Optional[Any] = None,\n        operation_name: Optional[str] = None,\n        allowed_operation_types: Optional[Iterable[OperationType]] = None,\n    ) -> ExecutionResult:\n        if allowed_operation_types is None:\n            allowed_operation_types = DEFAULT_ALLOWED_OPERATION_TYPES\n\n        execution_context = ExecutionContext(\n            query=query,\n            schema=self,\n            context=context_value,\n            root_value=root_value,\n            variables=variable_values,\n            provided_operation_name=operation_name,\n        )\n\n        result = execute_sync(\n            self._schema,\n            query,\n            extensions=list(self.extensions) + [DirectivesExtensionSync],\n            execution_context_class=self.execution_context_class,\n            execution_context=execution_context,\n            allowed_operation_types=allowed_operation_types,\n        )\n\n        if result.errors:\n            self.process_errors(result.errors, execution_context=execution_context)\n\n        return result\n\n    async def subscribe(\n        self,\n        query: str,\n        variable_values: Optional[Dict[str, Any]] = None,\n        context_value: Optional[Any] = None,\n        root_value: Optional[Any] = None,\n        operation_name: Optional[str] = None,\n    ):\n        return await subscribe(\n            self._schema,\n            parse(query),\n            root_value=root_value,\n            context_value=context_value,\n            variable_values=variable_values,\n            operation_name=operation_name,\n        )\n\n    def as_str(self) -> str:\n        return print_schema(self)\n\n    __str__ = as_str\n\n    def introspect(self) -> Dict[str, Any]:\n        \"\"\"Return the introspection query result for the current schema\n\n        Raises:\n            ValueError: If the introspection query fails due to an invalid schema\n        \"\"\"\n        introspection = self.execute_sync(get_introspection_query())\n        if introspection.errors or not introspection.data:\n            raise ValueError(f\"Invalid Schema. Errors {introspection.errors!r}\")\n\n        return introspection.data\n", "path": "strawberry/schema/schema.py"}], "after_files": [{"content": "from functools import lru_cache\nfrom typing import Any, Dict, Iterable, List, Optional, Sequence, Type, Union\n\nfrom graphql import (\n    ExecutionContext as GraphQLExecutionContext,\n    GraphQLNamedType,\n    GraphQLNonNull,\n    GraphQLSchema,\n    get_introspection_query,\n    parse,\n    validate_schema,\n)\nfrom graphql.subscription import subscribe\nfrom graphql.type.directives import specified_directives\n\nfrom strawberry.custom_scalar import ScalarDefinition, ScalarWrapper\nfrom strawberry.directive import StrawberryDirective\nfrom strawberry.enum import EnumDefinition\nfrom strawberry.extensions import Extension\nfrom strawberry.extensions.directives import (\n    DirectivesExtension,\n    DirectivesExtensionSync,\n)\nfrom strawberry.field import StrawberryField\nfrom strawberry.schema.schema_converter import GraphQLCoreConverter\nfrom strawberry.schema.types.scalar import DEFAULT_SCALAR_REGISTRY\nfrom strawberry.types import ExecutionContext, ExecutionResult\nfrom strawberry.types.graphql import OperationType\nfrom strawberry.types.types import TypeDefinition\nfrom strawberry.union import StrawberryUnion\n\nfrom ..printer import print_schema\nfrom .base import BaseSchema\nfrom .config import StrawberryConfig\nfrom .execute import execute, execute_sync\n\n\nDEFAULT_ALLOWED_OPERATION_TYPES = {\n    OperationType.QUERY,\n    OperationType.MUTATION,\n    OperationType.SUBSCRIPTION,\n}\n\n\nclass Schema(BaseSchema):\n    def __init__(\n        self,\n        # TODO: can we make sure we only allow to pass something that has been decorated?\n        query: Type,\n        mutation: Optional[Type] = None,\n        subscription: Optional[Type] = None,\n        directives: Sequence[StrawberryDirective] = (),\n        types=(),\n        extensions: Sequence[Union[Type[Extension], Extension]] = (),\n        execution_context_class: Optional[Type[GraphQLExecutionContext]] = None,\n        config: Optional[StrawberryConfig] = None,\n        scalar_overrides: Optional[\n            Dict[object, Union[ScalarWrapper, ScalarDefinition]]\n        ] = None,\n    ):\n        self.extensions = extensions\n        self.execution_context_class = execution_context_class\n        self.config = config or StrawberryConfig()\n\n        scalar_registry: Dict[object, Union[ScalarWrapper, ScalarDefinition]] = {\n            **DEFAULT_SCALAR_REGISTRY\n        }\n        if scalar_overrides:\n            scalar_registry.update(scalar_overrides)\n\n        self.schema_converter = GraphQLCoreConverter(self.config, scalar_registry)\n        self.directives = directives\n\n        query_type = self.schema_converter.from_object(query._type_definition)\n        mutation_type = (\n            self.schema_converter.from_object(mutation._type_definition)\n            if mutation\n            else None\n        )\n        subscription_type = (\n            self.schema_converter.from_object(subscription._type_definition)\n            if subscription\n            else None\n        )\n\n        graphql_directives = tuple(\n            self.schema_converter.from_directive(directive) for directive in directives\n        )\n\n        graphql_types = []\n        for type_ in types:\n            graphql_type = self.schema_converter.from_maybe_optional(type_)\n            if isinstance(graphql_type, GraphQLNonNull):\n                graphql_type = graphql_type.of_type\n            if not isinstance(graphql_type, GraphQLNamedType):\n                raise TypeError(f\"{graphql_type} is not a named GraphQL Type\")\n            graphql_types.append(graphql_type)\n\n        self._schema = GraphQLSchema(\n            query=query_type,\n            mutation=mutation_type,\n            subscription=subscription_type if subscription else None,\n            directives=specified_directives + graphql_directives,\n            types=graphql_types,\n            extensions={\n                GraphQLCoreConverter.DEFINITION_BACKREF: self,\n            },\n        )\n\n        # attach our schema to the GraphQL schema instance\n        self._schema._strawberry_schema = self  # type: ignore\n\n        # Validate schema early because we want developers to know about\n        # possible issues as soon as possible\n        errors = validate_schema(self._schema)\n        if errors:\n            formatted_errors = \"\\n\\n\".join(f\"\u274c {error.message}\" for error in errors)\n            raise ValueError(f\"Invalid Schema. Errors:\\n\\n{formatted_errors}\")\n\n        self.query = self.schema_converter.type_map[query_type.name]\n\n    def get_extensions(\n        self, sync: bool = False\n    ) -> List[Union[Type[Extension], Extension]]:\n        extensions = list(self.extensions)\n\n        if self.directives:\n            extensions.append(DirectivesExtensionSync if sync else DirectivesExtension)\n\n        return extensions\n\n    @lru_cache()\n    def get_type_by_name(  # type: ignore  # lru_cache makes mypy complain\n        self, name: str\n    ) -> Optional[\n        Union[TypeDefinition, ScalarDefinition, EnumDefinition, StrawberryUnion]\n    ]:\n        # TODO: respect auto_camel_case\n        if name in self.schema_converter.type_map:\n            return self.schema_converter.type_map[name].definition\n\n        return None\n\n    def get_field_for_type(\n        self, field_name: str, type_name: str\n    ) -> Optional[StrawberryField]:\n        type_ = self.get_type_by_name(type_name)\n\n        if not type_:\n            return None  # pragma: no cover\n\n        assert isinstance(type_, TypeDefinition)\n\n        return next(\n            (\n                field\n                for field in type_.fields\n                if self.config.name_converter.get_graphql_name(field) == field_name\n            ),\n            None,\n        )\n\n    @lru_cache()\n    def get_directive_by_name(self, graphql_name: str) -> Optional[StrawberryDirective]:\n        return next(\n            (\n                directive\n                for directive in self.directives\n                if self.config.name_converter.from_directive(directive) == graphql_name\n            ),\n            None,\n        )\n\n    async def execute(\n        self,\n        query: str,\n        variable_values: Optional[Dict[str, Any]] = None,\n        context_value: Optional[Any] = None,\n        root_value: Optional[Any] = None,\n        operation_name: Optional[str] = None,\n        allowed_operation_types: Optional[Iterable[OperationType]] = None,\n    ) -> ExecutionResult:\n        if allowed_operation_types is None:\n            allowed_operation_types = DEFAULT_ALLOWED_OPERATION_TYPES\n\n        # Create execution context\n        execution_context = ExecutionContext(\n            query=query,\n            schema=self,\n            context=context_value,\n            root_value=root_value,\n            variables=variable_values,\n            provided_operation_name=operation_name,\n        )\n\n        result = await execute(\n            self._schema,\n            query,\n            extensions=self.get_extensions(),\n            execution_context_class=self.execution_context_class,\n            execution_context=execution_context,\n            allowed_operation_types=allowed_operation_types,\n        )\n\n        if result.errors:\n            self.process_errors(result.errors, execution_context=execution_context)\n\n        return result\n\n    def execute_sync(\n        self,\n        query: str,\n        variable_values: Optional[Dict[str, Any]] = None,\n        context_value: Optional[Any] = None,\n        root_value: Optional[Any] = None,\n        operation_name: Optional[str] = None,\n        allowed_operation_types: Optional[Iterable[OperationType]] = None,\n    ) -> ExecutionResult:\n        if allowed_operation_types is None:\n            allowed_operation_types = DEFAULT_ALLOWED_OPERATION_TYPES\n\n        execution_context = ExecutionContext(\n            query=query,\n            schema=self,\n            context=context_value,\n            root_value=root_value,\n            variables=variable_values,\n            provided_operation_name=operation_name,\n        )\n\n        result = execute_sync(\n            self._schema,\n            query,\n            extensions=self.get_extensions(sync=True),\n            execution_context_class=self.execution_context_class,\n            execution_context=execution_context,\n            allowed_operation_types=allowed_operation_types,\n        )\n\n        if result.errors:\n            self.process_errors(result.errors, execution_context=execution_context)\n\n        return result\n\n    async def subscribe(\n        self,\n        query: str,\n        variable_values: Optional[Dict[str, Any]] = None,\n        context_value: Optional[Any] = None,\n        root_value: Optional[Any] = None,\n        operation_name: Optional[str] = None,\n    ):\n        return await subscribe(\n            self._schema,\n            parse(query),\n            root_value=root_value,\n            context_value=context_value,\n            variable_values=variable_values,\n            operation_name=operation_name,\n        )\n\n    def as_str(self) -> str:\n        return print_schema(self)\n\n    __str__ = as_str\n\n    def introspect(self) -> Dict[str, Any]:\n        \"\"\"Return the introspection query result for the current schema\n\n        Raises:\n            ValueError: If the introspection query fails due to an invalid schema\n        \"\"\"\n        introspection = self.execute_sync(get_introspection_query())\n        if introspection.errors or not introspection.data:\n            raise ValueError(f\"Invalid Schema. Errors {introspection.errors!r}\")\n\n        return introspection.data\n", "path": "strawberry/schema/schema.py"}]}