DSL meta fields implementation

docs/advanced/dsl_module.rst

@@ -338,6 +338,16 @@ this can be written in a concise manner::
         DSLInlineFragment().on(ds.Human).select(ds.Human.homePlanet)
     )
 
+Meta-fields
+^^^^^^^^^^^
+
+To define meta-fields (:code:`__typename`, :code:`__schema` and :code:`__type`),
+you can use the :class:`DSLMetaField <gql.dsl.DSLMetaField>` class::
+
+    query = ds.Query.hero.select(
+        ds.Character.name,
+        DSLMetaField("__typename")
+    )
 
 Executable examples
 -------------------

gql/dsl.py

@@ -25,6 +25,7 @@
     GraphQLNonNull,
     GraphQLObjectType,
     GraphQLSchema,
+    GraphQLString,
     GraphQLWrappingType,
     InlineFragmentNode,
     IntValueNode,
@@ -46,6 +47,7 @@
     VariableDefinitionNode,
     VariableNode,
     assert_named_type,
+    introspection_types,
     is_enum_type,
     is_input_object_type,
     is_leaf_type,
@@ -301,6 +303,7 @@ class DSLExecutable(ABC):
 
     variable_definitions: "DSLVariableDefinitions"
     name: Optional[str]
+    selection_set: SelectionSetNode
 
     @property
     @abstractmethod
@@ -349,11 +352,31 @@ def __init__(
                         f"Received type: {type(field)}"
                     )
                 )
+            valid_type = False
             if isinstance(self, DSLOperation):
-                assert field.type_name.upper() == self.operation_type.name, (
-                    f"Invalid root field for operation {self.operation_type.name}.\n"
-                    f"Received: {field.type_name}"
-                )
+                operation_name = self.operation_type.name
+                if isinstance(field, DSLMetaField):
+                    if field.name in ["__schema", "__type"]:
+                        valid_type = operation_name == "QUERY"
+                    if field.name == "__typename":
+                        valid_type = operation_name != "SUBSCRIPTION"
+                else:
+                    valid_type = field.parent_type.name.upper() == operation_name
+
+            else:  # Fragments
+                if isinstance(field, DSLMetaField):
+                    valid_type = field.name == "__typename"
+
+            if not valid_type:
+                if isinstance(self, DSLOperation):
+                    error_msg = (
+                        "Invalid root field for operation "
+                        f"{self.operation_type.name}"
+                    )
+                else:
+                    error_msg = f"Invalid field for fragment {self.name}"
+
+                raise AssertionError(f"{error_msg}: {field!r}")
 
         self.selection_set = SelectionSetNode(
             selections=FrozenList(DSLSelectable.get_ast_fields(all_fields))
@@ -610,6 +633,11 @@ def select(
             fields, fields_with_alias
         )
 
+        # Check that we don't receive an invalid meta-field
+        for field in added_fields:
+            if isinstance(field, DSLMetaField) and field.name != "__typename":
+                raise AssertionError(f"Invalid field for {self!r}: {field!r}")
+
         # Get a list of AST Nodes for each added field
         added_selections: List[
             Union[FieldNode, InlineFragmentNode, FragmentSpreadNode]
@@ -668,8 +696,8 @@ class DSLField(DSLSelectableWithAlias, DSLSelector):
     def __init__(
         self,
         name: str,
-        graphql_type: Union[GraphQLObjectType, GraphQLInterfaceType],
-        graphql_field: GraphQLField,
+        parent_type: Union[GraphQLObjectType, GraphQLInterfaceType],
+        field: GraphQLField,
     ):
         """Initialize the DSLField.
 
@@ -678,15 +706,21 @@ def __init__(
             Use attributes of the :class:`DSLType` instead.
 
         :param name: the name of the field
-        :param graphql_type: the GraphQL type definition from the schema
-        :param graphql_field: the GraphQL field definition from the schema
+        :param parent_type: the GraphQL type definition from the schema of the
+                            parent type of the field
+        :param field: the GraphQL field definition from the schema
         """
         DSLSelector.__init__(self)
-        self._type = graphql_type
-        self.field = graphql_field
+        self.parent_type = parent_type
+        self.field = field
         self.ast_field = FieldNode(name=NameNode(value=name), arguments=FrozenList())
         log.debug(f"Creating {self!r}")
 
+    @property
+    def name(self):
+        """:meta private:"""
+        return self.ast_field.name.value
+
     def __call__(self, **kwargs) -> "DSLField":
         return self.args(**kwargs)
 
@@ -750,16 +784,49 @@ def select(
 
         return self
 
-    @property
-    def type_name(self):
-        """:meta private:"""
-        return self._type.name
-
     def __repr__(self) -> str:
-        return (
-            f"<{self.__class__.__name__} {self._type.name}"
-            f"::{self.ast_field.name.value}>"
-        )
+        return f"<{self.__class__.__name__} {self.parent_type.name}" f"::{self.name}>"
+
+
+class DSLMetaField(DSLField):
+    """DSLMetaField represents a GraphQL meta-field for the DSL code.
+
+    meta-fields are reserved field in the GraphQL type system prefixed with
+    "__" two underscores and used for introspection.
+    """
+
+    meta_type = GraphQLObjectType(
+        "meta-field",
+        fields={
+            "__typename": GraphQLField(GraphQLString),
+            "__schema": GraphQLField(
+                cast(GraphQLObjectType, introspection_types["__Schema"])
+            ),
+            "__type": GraphQLField(
+                cast(GraphQLObjectType, introspection_types["__Type"]),
+                args={"name": GraphQLArgument(type_=GraphQLNonNull(GraphQLString))},
+            ),
+        },
+    )
+
+    def __init__(self, name: str):
+        """Initialize the meta-field.
+
+        :param name: the name between __typename, __schema or __type
+        """
+
+        try:
+            field = self.meta_type.fields[name]
+        except KeyError:
+            raise AssertionError(f'Invalid meta-field "{name}"')
+
+        super().__init__(name, self.meta_type, field)
+
+    def alias(self, alias: str) -> "DSLSelectableWithAlias":
+        """
+        :meta private:
+        """
+        pass
 
 
 class DSLInlineFragment(DSLSelectable, DSLSelector):

gql/utilities/__init__.py

@@ -1,13 +1,15 @@
+from .get_introspection_query_ast import get_introspection_query_ast
 from .parse_result import parse_result
 from .serialize_variable_values import serialize_value, serialize_variable_values
 from .update_schema_enum import update_schema_enum
 from .update_schema_scalars import update_schema_scalar, update_schema_scalars
 
 __all__ = [
-    "update_schema_scalars",
-    "update_schema_scalar",
-    "update_schema_enum",
     "parse_result",
+    "get_introspection_query_ast",
     "serialize_variable_values",
     "serialize_value",
+    "update_schema_enum",
+    "update_schema_scalars",
+    "update_schema_scalar",
 ]

gql/utilities/get_introspection_query_ast.py

@@ -0,0 +1,123 @@
+from itertools import repeat
+
+from graphql import DocumentNode, GraphQLSchema
+
+from gql.dsl import DSLFragment, DSLMetaField, DSLQuery, DSLSchema, dsl_gql
+
+
+def get_introspection_query_ast(
+    descriptions: bool = True,
+    specified_by_url: bool = False,
+    directive_is_repeatable: bool = False,
+    schema_description: bool = False,
+    type_recursion_level: int = 7,
+) -> DocumentNode:
+    """Get a query for introspection as a document using the DSL module.
+
+    Equivalent to the get_introspection_query function from graphql-core
+    but using the DSL module and allowing to select the recursion level.
+
+    Optionally, you can exclude descriptions, include specification URLs,
+    include repeatability of directives, and specify whether to include
+    the schema description as well.
+    """
+
+    ds = DSLSchema(GraphQLSchema())
+
+    fragment_FullType = DSLFragment("FullType").on(ds.__Type)
+    fragment_InputValue = DSLFragment("InputValue").on(ds.__InputValue)
+    fragment_TypeRef = DSLFragment("TypeRef").on(ds.__Type)
+
+    schema = DSLMetaField("__schema")
+
+    if descriptions and schema_description:
+        schema.select(ds.__Schema.description)
+
+    schema.select(
+        ds.__Schema.queryType.select(ds.__Type.name),
+        ds.__Schema.mutationType.select(ds.__Type.name),
+        ds.__Schema.subscriptionType.select(ds.__Type.name),
+    )
+
+    schema.select(ds.__Schema.types.select(fragment_FullType))
+
+    directives = ds.__Schema.directives.select(ds.__Directive.name)
+
+    if descriptions:
+        directives.select(ds.__Directive.description)
+    if directive_is_repeatable:
+        directives.select(ds.__Directive.isRepeatable)
+    directives.select(
+        ds.__Directive.locations, ds.__Directive.args.select(fragment_InputValue),
+    )
+
+    schema.select(directives)
+
+    fragment_FullType.select(
+        ds.__Type.kind, ds.__Type.name,
+    )
+    if descriptions:
+        fragment_FullType.select(ds.__Type.description)
+    if specified_by_url:
+        fragment_FullType.select(ds.__Type.specifiedByUrl)
+
+    fields = ds.__Type.fields(includeDeprecated=True).select(ds.__Field.name)
+
+    if descriptions:
+        fields.select(ds.__Field.description)
+
+    fields.select(
+        ds.__Field.args.select(fragment_InputValue),
+        ds.__Field.type.select(fragment_TypeRef),
+        ds.__Field.isDeprecated,
+        ds.__Field.deprecationReason,
+    )
+
+    enum_values = ds.__Type.enumValues(includeDeprecated=True).select(
+        ds.__EnumValue.name
+    )
+
+    if descriptions:
+        enum_values.select(ds.__EnumValue.description)
+
+    enum_values.select(
+        ds.__EnumValue.isDeprecated, ds.__EnumValue.deprecationReason,
+    )
+
+    fragment_FullType.select(
+        fields,
+        ds.__Type.inputFields.select(fragment_InputValue),
+        ds.__Type.interfaces.select(fragment_TypeRef),
+        enum_values,
+        ds.__Type.possibleTypes.select(fragment_TypeRef),
+    )
+
+    fragment_InputValue.select(ds.__InputValue.name)
+
+    if descriptions:
+        fragment_InputValue.select(ds.__InputValue.description)
+
+    fragment_InputValue.select(
+        ds.__InputValue.type.select(fragment_TypeRef), ds.__InputValue.defaultValue,
+    )
+
+    fragment_TypeRef.select(
+        ds.__Type.kind, ds.__Type.name,
+    )
+
+    if type_recursion_level >= 1:
+        current_field = ds.__Type.ofType.select(ds.__Type.kind, ds.__Type.name)
+        fragment_TypeRef.select(current_field)
+
+        for _ in repeat(None, type_recursion_level - 1):
+            new_oftype = ds.__Type.ofType.select(ds.__Type.kind, ds.__Type.name)
+            current_field.select(new_oftype)
+            current_field = new_oftype
+
+    query = DSLQuery(schema)
+
+    query.name = "IntrospectionQuery"
+
+    dsl_query = dsl_gql(query, fragment_FullType, fragment_InputValue, fragment_TypeRef)
+
+    return dsl_query