We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add a warning when using `@strawberry.federation.type` but not using `strawberry.federation.Schema`
This should make developer life easier if they forget to use the federation schema :)

See: https://discord.com/channels/689806334337482765/773519351423827978/950480727630303232

--- 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 __future__ import annotations
2 
3 from functools import lru_cache
4 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Type, Union, cast
5 
6 from graphql import (
7     GraphQLNamedType,
8     GraphQLNonNull,
9     GraphQLSchema,
10     get_introspection_query,
11     parse,
12     validate_schema,
13 )
14 from graphql.subscription import subscribe
15 from graphql.type.directives import specified_directives
16 
17 from strawberry.annotation import StrawberryAnnotation
18 from strawberry.extensions.directives import (
19     DirectivesExtension,
20     DirectivesExtensionSync,
21 )
22 from strawberry.schema.schema_converter import GraphQLCoreConverter
23 from strawberry.schema.types.scalar import DEFAULT_SCALAR_REGISTRY
24 from strawberry.types import ExecutionContext
25 from strawberry.types.graphql import OperationType
26 from strawberry.types.types import TypeDefinition
27 
28 from ..printer import print_schema
29 from . import compat
30 from .base import BaseSchema
31 from .config import StrawberryConfig
32 from .execute import execute, execute_sync
33 
34 if TYPE_CHECKING:
35     from graphql import ExecutionContext as GraphQLExecutionContext
36 
37     from strawberry.custom_scalar import ScalarDefinition, ScalarWrapper
38     from strawberry.directive import StrawberryDirective
39     from strawberry.enum import EnumDefinition
40     from strawberry.extensions import SchemaExtension
41     from strawberry.field import StrawberryField
42     from strawberry.types import ExecutionResult
43     from strawberry.union import StrawberryUnion
44 
45 DEFAULT_ALLOWED_OPERATION_TYPES = {
46     OperationType.QUERY,
47     OperationType.MUTATION,
48     OperationType.SUBSCRIPTION,
49 }
50 
51 
52 class Schema(BaseSchema):
53     def __init__(
54         self,
55         # TODO: can we make sure we only allow to pass
56         # something that has been decorated?
57         query: Type,
58         mutation: Optional[Type] = None,
59         subscription: Optional[Type] = None,
60         directives: Iterable[StrawberryDirective] = (),
61         types=(),
62         extensions: Iterable[Union[Type[SchemaExtension], SchemaExtension]] = (),
63         execution_context_class: Optional[Type[GraphQLExecutionContext]] = None,
64         config: Optional[StrawberryConfig] = None,
65         scalar_overrides: Optional[
66             Dict[object, Union[Type, ScalarWrapper, ScalarDefinition]]
67         ] = None,
68         schema_directives: Iterable[object] = (),
69     ):
70         self.query = query
71         self.mutation = mutation
72         self.subscription = subscription
73 
74         self.extensions = extensions
75         self.execution_context_class = execution_context_class
76         self.config = config or StrawberryConfig()
77 
78         SCALAR_OVERRIDES_DICT_TYPE = Dict[
79             object, Union["ScalarWrapper", "ScalarDefinition"]
80         ]
81 
82         scalar_registry: SCALAR_OVERRIDES_DICT_TYPE = {**DEFAULT_SCALAR_REGISTRY}
83         if scalar_overrides:
84             # TODO: check that the overrides are valid
85             scalar_registry.update(cast(SCALAR_OVERRIDES_DICT_TYPE, scalar_overrides))
86 
87         self.schema_converter = GraphQLCoreConverter(self.config, scalar_registry)
88         self.directives = directives
89         self.schema_directives = list(schema_directives)
90 
91         query_type = self.schema_converter.from_object(query._type_definition)
92         mutation_type = (
93             self.schema_converter.from_object(mutation._type_definition)
94             if mutation
95             else None
96         )
97         subscription_type = (
98             self.schema_converter.from_object(subscription._type_definition)
99             if subscription
100             else None
101         )
102 
103         graphql_directives = [
104             self.schema_converter.from_directive(directive) for directive in directives
105         ]
106 
107         graphql_types = []
108         for type_ in types:
109             if compat.is_schema_directive(type_):
110                 graphql_directives.append(
111                     self.schema_converter.from_schema_directive(type_)
112                 )
113             else:
114                 if hasattr(type_, "_type_definition"):
115                     if type_._type_definition.is_generic:
116                         type_ = StrawberryAnnotation(type_).resolve()
117                 graphql_type = self.schema_converter.from_maybe_optional(type_)
118                 if isinstance(graphql_type, GraphQLNonNull):
119                     graphql_type = graphql_type.of_type
120                 if not isinstance(graphql_type, GraphQLNamedType):
121                     raise TypeError(f"{graphql_type} is not a named GraphQL Type")
122                 graphql_types.append(graphql_type)
123 
124         try:
125             self._schema = GraphQLSchema(
126                 query=query_type,
127                 mutation=mutation_type,
128                 subscription=subscription_type if subscription else None,
129                 directives=specified_directives + tuple(graphql_directives),
130                 types=graphql_types,
131                 extensions={
132                     GraphQLCoreConverter.DEFINITION_BACKREF: self,
133                 },
134             )
135 
136         except TypeError as error:
137             # GraphQL core throws a TypeError if there's any exception raised
138             # during the schema creation, so we check if the cause was a
139             # StrawberryError and raise it instead if that's the case.
140 
141             from strawberry.exceptions import StrawberryException
142 
143             if isinstance(error.__cause__, StrawberryException):
144                 raise error.__cause__ from None
145 
146             raise
147 
148         # attach our schema to the GraphQL schema instance
149         self._schema._strawberry_schema = self  # type: ignore
150 
151         # Validate schema early because we want developers to know about
152         # possible issues as soon as possible
153         errors = validate_schema(self._schema)
154         if errors:
155             formatted_errors = "\n\n".join(f"❌ {error.message}" for error in errors)
156             raise ValueError(f"Invalid Schema. Errors:\n\n{formatted_errors}")
157 
158     def get_extensions(
159         self, sync: bool = False
160     ) -> List[Union[Type[SchemaExtension], SchemaExtension]]:
161         extensions = list(self.extensions)
162 
163         if self.directives:
164             extensions.append(DirectivesExtensionSync if sync else DirectivesExtension)
165 
166         return extensions
167 
168     @lru_cache()
169     def get_type_by_name(
170         self, name: str
171     ) -> Optional[
172         Union[TypeDefinition, ScalarDefinition, EnumDefinition, StrawberryUnion]
173     ]:
174         # TODO: respect auto_camel_case
175         if name in self.schema_converter.type_map:
176             return self.schema_converter.type_map[name].definition
177 
178         return None
179 
180     def get_field_for_type(
181         self, field_name: str, type_name: str
182     ) -> Optional[StrawberryField]:
183         type_ = self.get_type_by_name(type_name)
184 
185         if not type_:
186             return None  # pragma: no cover
187 
188         assert isinstance(type_, TypeDefinition)
189 
190         return next(
191             (
192                 field
193                 for field in type_.fields
194                 if self.config.name_converter.get_graphql_name(field) == field_name
195             ),
196             None,
197         )
198 
199     @lru_cache()
200     def get_directive_by_name(self, graphql_name: str) -> Optional[StrawberryDirective]:
201         return next(
202             (
203                 directive
204                 for directive in self.directives
205                 if self.config.name_converter.from_directive(directive) == graphql_name
206             ),
207             None,
208         )
209 
210     async def execute(
211         self,
212         query: Optional[str],
213         variable_values: Optional[Dict[str, Any]] = None,
214         context_value: Optional[Any] = None,
215         root_value: Optional[Any] = None,
216         operation_name: Optional[str] = None,
217         allowed_operation_types: Optional[Iterable[OperationType]] = None,
218     ) -> ExecutionResult:
219         if allowed_operation_types is None:
220             allowed_operation_types = DEFAULT_ALLOWED_OPERATION_TYPES
221 
222         # Create execution context
223         execution_context = ExecutionContext(
224             query=query,
225             schema=self,
226             context=context_value,
227             root_value=root_value,
228             variables=variable_values,
229             provided_operation_name=operation_name,
230         )
231 
232         result = await execute(
233             self._schema,
234             extensions=self.get_extensions(),
235             execution_context_class=self.execution_context_class,
236             execution_context=execution_context,
237             allowed_operation_types=allowed_operation_types,
238             process_errors=self.process_errors,
239         )
240 
241         return result
242 
243     def execute_sync(
244         self,
245         query: Optional[str],
246         variable_values: Optional[Dict[str, Any]] = None,
247         context_value: Optional[Any] = None,
248         root_value: Optional[Any] = None,
249         operation_name: Optional[str] = None,
250         allowed_operation_types: Optional[Iterable[OperationType]] = None,
251     ) -> ExecutionResult:
252         if allowed_operation_types is None:
253             allowed_operation_types = DEFAULT_ALLOWED_OPERATION_TYPES
254 
255         execution_context = ExecutionContext(
256             query=query,
257             schema=self,
258             context=context_value,
259             root_value=root_value,
260             variables=variable_values,
261             provided_operation_name=operation_name,
262         )
263 
264         result = execute_sync(
265             self._schema,
266             extensions=self.get_extensions(sync=True),
267             execution_context_class=self.execution_context_class,
268             execution_context=execution_context,
269             allowed_operation_types=allowed_operation_types,
270             process_errors=self.process_errors,
271         )
272 
273         return result
274 
275     async def subscribe(
276         self,
277         # TODO: make this optional when we support extensions
278         query: str,
279         variable_values: Optional[Dict[str, Any]] = None,
280         context_value: Optional[Any] = None,
281         root_value: Optional[Any] = None,
282         operation_name: Optional[str] = None,
283     ):
284         return await subscribe(
285             self._schema,
286             parse(query),
287             root_value=root_value,
288             context_value=context_value,
289             variable_values=variable_values,
290             operation_name=operation_name,
291         )
292 
293     def as_str(self) -> str:
294         return print_schema(self)
295 
296     __str__ = as_str
297 
298     def introspect(self) -> Dict[str, Any]:
299         """Return the introspection query result for the current schema
300 
301         Raises:
302             ValueError: If the introspection query fails due to an invalid schema
303         """
304         introspection = self.execute_sync(get_introspection_query())
305         if introspection.errors or not introspection.data:
306             raise ValueError(f"Invalid Schema. Errors {introspection.errors!r}")
307 
308         return introspection.data
309
```
--- 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,6 @@
 from __future__ import annotations
 
+import warnings
 from functools import lru_cache
 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Type, Union, cast
 
@@ -148,6 +149,8 @@
         # attach our schema to the GraphQL schema instance
         self._schema._strawberry_schema = self  # type: ignore
 
+        self._warn_for_federation_directives()
+
         # Validate schema early because we want developers to know about
         # possible issues as soon as possible
         errors = validate_schema(self._schema)
@@ -290,6 +293,25 @@
             operation_name=operation_name,
         )
 
+    def _warn_for_federation_directives(self):
+        """Raises a warning if the schema has any federation directives."""
+        from strawberry.federation.schema_directives import FederationDirective
+
+        if any(
+            type_
+            for type_ in self.schema_converter.type_map.values()
+            if any(
+                directive
+                for directive in (type_.definition.directives or [])
+                if isinstance(directive, FederationDirective)
+            )
+        ):
+            warnings.warn(
+                "Federation directive found in schema. "
+                "Should use strawberry.federation.Schema instead.",
+                UserWarning,
+            )
+
     def as_str(self) -> str:
         return print_schema(self)

{"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,6 @@\n from __future__ import annotations\n \n+import warnings\n from functools import lru_cache\n from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Type, Union, cast\n \n@@ -148,6 +149,8 @@\n         # attach our schema to the GraphQL schema instance\n         self._schema._strawberry_schema = self  # type: ignore\n \n+        self._warn_for_federation_directives()\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@@ -290,6 +293,25 @@\n             operation_name=operation_name,\n         )\n \n+    def _warn_for_federation_directives(self):\n+        \"\"\"Raises a warning if the schema has any federation directives.\"\"\"\n+        from strawberry.federation.schema_directives import FederationDirective\n+\n+        if any(\n+            type_\n+            for type_ in self.schema_converter.type_map.values()\n+            if any(\n+                directive\n+                for directive in (type_.definition.directives or [])\n+                if isinstance(directive, FederationDirective)\n+            )\n+        ):\n+            warnings.warn(\n+                \"Federation directive found in schema. \"\n+                \"Should use strawberry.federation.Schema instead.\",\n+                UserWarning,\n+            )\n+\n     def as_str(self) -> str:\n         return print_schema(self)\n", "issue": "Add a warning when using `@strawberry.federation.type` but not using `strawberry.federation.Schema`\nThis should make developer life easier if they forget to use the federation schema :)\r\n\r\nSee: https://discord.com/channels/689806334337482765/773519351423827978/950480727630303232\n", "before_files": [{"content": "from __future__ import annotations\n\nfrom functools import lru_cache\nfrom typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Type, Union, cast\n\nfrom graphql import (\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.annotation import StrawberryAnnotation\nfrom strawberry.extensions.directives import (\n    DirectivesExtension,\n    DirectivesExtensionSync,\n)\nfrom strawberry.schema.schema_converter import GraphQLCoreConverter\nfrom strawberry.schema.types.scalar import DEFAULT_SCALAR_REGISTRY\nfrom strawberry.types import ExecutionContext\nfrom strawberry.types.graphql import OperationType\nfrom strawberry.types.types import TypeDefinition\n\nfrom ..printer import print_schema\nfrom . import compat\nfrom .base import BaseSchema\nfrom .config import StrawberryConfig\nfrom .execute import execute, execute_sync\n\nif TYPE_CHECKING:\n    from graphql import ExecutionContext as GraphQLExecutionContext\n\n    from strawberry.custom_scalar import ScalarDefinition, ScalarWrapper\n    from strawberry.directive import StrawberryDirective\n    from strawberry.enum import EnumDefinition\n    from strawberry.extensions import SchemaExtension\n    from strawberry.field import StrawberryField\n    from strawberry.types import ExecutionResult\n    from strawberry.union import StrawberryUnion\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\n        # something that has been decorated?\n        query: Type,\n        mutation: Optional[Type] = None,\n        subscription: Optional[Type] = None,\n        directives: Iterable[StrawberryDirective] = (),\n        types=(),\n        extensions: Iterable[Union[Type[SchemaExtension], SchemaExtension]] = (),\n        execution_context_class: Optional[Type[GraphQLExecutionContext]] = None,\n        config: Optional[StrawberryConfig] = None,\n        scalar_overrides: Optional[\n            Dict[object, Union[Type, ScalarWrapper, ScalarDefinition]]\n        ] = None,\n        schema_directives: Iterable[object] = (),\n    ):\n        self.query = query\n        self.mutation = mutation\n        self.subscription = subscription\n\n        self.extensions = extensions\n        self.execution_context_class = execution_context_class\n        self.config = config or StrawberryConfig()\n\n        SCALAR_OVERRIDES_DICT_TYPE = Dict[\n            object, Union[\"ScalarWrapper\", \"ScalarDefinition\"]\n        ]\n\n        scalar_registry: SCALAR_OVERRIDES_DICT_TYPE = {**DEFAULT_SCALAR_REGISTRY}\n        if scalar_overrides:\n            # TODO: check that the overrides are valid\n            scalar_registry.update(cast(SCALAR_OVERRIDES_DICT_TYPE, scalar_overrides))\n\n        self.schema_converter = GraphQLCoreConverter(self.config, scalar_registry)\n        self.directives = directives\n        self.schema_directives = list(schema_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 = [\n            self.schema_converter.from_directive(directive) for directive in directives\n        ]\n\n        graphql_types = []\n        for type_ in types:\n            if compat.is_schema_directive(type_):\n                graphql_directives.append(\n                    self.schema_converter.from_schema_directive(type_)\n                )\n            else:\n                if hasattr(type_, \"_type_definition\"):\n                    if type_._type_definition.is_generic:\n                        type_ = StrawberryAnnotation(type_).resolve()\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        try:\n            self._schema = GraphQLSchema(\n                query=query_type,\n                mutation=mutation_type,\n                subscription=subscription_type if subscription else None,\n                directives=specified_directives + tuple(graphql_directives),\n                types=graphql_types,\n                extensions={\n                    GraphQLCoreConverter.DEFINITION_BACKREF: self,\n                },\n            )\n\n        except TypeError as error:\n            # GraphQL core throws a TypeError if there's any exception raised\n            # during the schema creation, so we check if the cause was a\n            # StrawberryError and raise it instead if that's the case.\n\n            from strawberry.exceptions import StrawberryException\n\n            if isinstance(error.__cause__, StrawberryException):\n                raise error.__cause__ from None\n\n            raise\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    def get_extensions(\n        self, sync: bool = False\n    ) -> List[Union[Type[SchemaExtension], SchemaExtension]]:\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(\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: Optional[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            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            process_errors=self.process_errors,\n        )\n\n        return result\n\n    def execute_sync(\n        self,\n        query: Optional[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            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            process_errors=self.process_errors,\n        )\n\n        return result\n\n    async def subscribe(\n        self,\n        # TODO: make this optional when we support extensions\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 __future__ import annotations\n\nimport warnings\nfrom functools import lru_cache\nfrom typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Type, Union, cast\n\nfrom graphql import (\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.annotation import StrawberryAnnotation\nfrom strawberry.extensions.directives import (\n    DirectivesExtension,\n    DirectivesExtensionSync,\n)\nfrom strawberry.schema.schema_converter import GraphQLCoreConverter\nfrom strawberry.schema.types.scalar import DEFAULT_SCALAR_REGISTRY\nfrom strawberry.types import ExecutionContext\nfrom strawberry.types.graphql import OperationType\nfrom strawberry.types.types import TypeDefinition\n\nfrom ..printer import print_schema\nfrom . import compat\nfrom .base import BaseSchema\nfrom .config import StrawberryConfig\nfrom .execute import execute, execute_sync\n\nif TYPE_CHECKING:\n    from graphql import ExecutionContext as GraphQLExecutionContext\n\n    from strawberry.custom_scalar import ScalarDefinition, ScalarWrapper\n    from strawberry.directive import StrawberryDirective\n    from strawberry.enum import EnumDefinition\n    from strawberry.extensions import SchemaExtension\n    from strawberry.field import StrawberryField\n    from strawberry.types import ExecutionResult\n    from strawberry.union import StrawberryUnion\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\n        # something that has been decorated?\n        query: Type,\n        mutation: Optional[Type] = None,\n        subscription: Optional[Type] = None,\n        directives: Iterable[StrawberryDirective] = (),\n        types=(),\n        extensions: Iterable[Union[Type[SchemaExtension], SchemaExtension]] = (),\n        execution_context_class: Optional[Type[GraphQLExecutionContext]] = None,\n        config: Optional[StrawberryConfig] = None,\n        scalar_overrides: Optional[\n            Dict[object, Union[Type, ScalarWrapper, ScalarDefinition]]\n        ] = None,\n        schema_directives: Iterable[object] = (),\n    ):\n        self.query = query\n        self.mutation = mutation\n        self.subscription = subscription\n\n        self.extensions = extensions\n        self.execution_context_class = execution_context_class\n        self.config = config or StrawberryConfig()\n\n        SCALAR_OVERRIDES_DICT_TYPE = Dict[\n            object, Union[\"ScalarWrapper\", \"ScalarDefinition\"]\n        ]\n\n        scalar_registry: SCALAR_OVERRIDES_DICT_TYPE = {**DEFAULT_SCALAR_REGISTRY}\n        if scalar_overrides:\n            # TODO: check that the overrides are valid\n            scalar_registry.update(cast(SCALAR_OVERRIDES_DICT_TYPE, scalar_overrides))\n\n        self.schema_converter = GraphQLCoreConverter(self.config, scalar_registry)\n        self.directives = directives\n        self.schema_directives = list(schema_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 = [\n            self.schema_converter.from_directive(directive) for directive in directives\n        ]\n\n        graphql_types = []\n        for type_ in types:\n            if compat.is_schema_directive(type_):\n                graphql_directives.append(\n                    self.schema_converter.from_schema_directive(type_)\n                )\n            else:\n                if hasattr(type_, \"_type_definition\"):\n                    if type_._type_definition.is_generic:\n                        type_ = StrawberryAnnotation(type_).resolve()\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        try:\n            self._schema = GraphQLSchema(\n                query=query_type,\n                mutation=mutation_type,\n                subscription=subscription_type if subscription else None,\n                directives=specified_directives + tuple(graphql_directives),\n                types=graphql_types,\n                extensions={\n                    GraphQLCoreConverter.DEFINITION_BACKREF: self,\n                },\n            )\n\n        except TypeError as error:\n            # GraphQL core throws a TypeError if there's any exception raised\n            # during the schema creation, so we check if the cause was a\n            # StrawberryError and raise it instead if that's the case.\n\n            from strawberry.exceptions import StrawberryException\n\n            if isinstance(error.__cause__, StrawberryException):\n                raise error.__cause__ from None\n\n            raise\n\n        # attach our schema to the GraphQL schema instance\n        self._schema._strawberry_schema = self  # type: ignore\n\n        self._warn_for_federation_directives()\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    def get_extensions(\n        self, sync: bool = False\n    ) -> List[Union[Type[SchemaExtension], SchemaExtension]]:\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(\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: Optional[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            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            process_errors=self.process_errors,\n        )\n\n        return result\n\n    def execute_sync(\n        self,\n        query: Optional[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            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            process_errors=self.process_errors,\n        )\n\n        return result\n\n    async def subscribe(\n        self,\n        # TODO: make this optional when we support extensions\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 _warn_for_federation_directives(self):\n        \"\"\"Raises a warning if the schema has any federation directives.\"\"\"\n        from strawberry.federation.schema_directives import FederationDirective\n\n        if any(\n            type_\n            for type_ in self.schema_converter.type_map.values()\n            if any(\n                directive\n                for directive in (type_.definition.directives or [])\n                if isinstance(directive, FederationDirective)\n            )\n        ):\n            warnings.warn(\n                \"Federation directive found in schema. \"\n                \"Should use strawberry.federation.Schema instead.\",\n                UserWarning,\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"}]}