Compose queries dynamically

Instead of providing the GraphQL queries as a Python String, it is also possible to create GraphQL queries dynamically. Using the DSL module, we can create a query using a Domain Specific Language which is created from the schema.

The following code:

ds = DSLSchema(StarWarsSchema)

query = dsl_gql(
    DSLQuery(
        ds.Query.hero.select(
            ds.Character.id,
            ds.Character.name,
            ds.Character.friends.select(ds.Character.name),
        )
    )
)

will generate a query equivalent to:

query = gql("""
    query {
      hero {
        id
        name
        friends {
          name
        }
      }
    }
""")

How to use

First generate the root using the DSLSchema:

ds = DSLSchema(client.schema)

Then use auto-generated attributes of the ds instance to get a root type (Query, Mutation or Subscription). This will generate a DSLType instance:

ds.Query

From this root type, you use auto-generated attributes to get a field. This will generate a DSLField instance:

ds.Query.hero

hero is a GraphQL object type and needs children fields. By default, there is no children fields selected. To select the fields that you want in your query, you use the select method.

To generate the children fields, we use the same method as above to auto-generate the fields from the ds instance (ie ds.Character.name is the field name of the type Character):

ds.Query.hero.select(ds.Character.name)

The select method return the same instance, so it is possible to chain the calls:

ds.Query.hero.select(ds.Character.name).select(ds.Character.id)

Or do it sequencially:

hero_query = ds.Query.hero

hero_query.select(ds.Character.name)
hero_query.select(ds.Character.id)

As you can select children fields of any object type, you can construct your complete query tree:

ds.Query.hero.select(
    ds.Character.id,
    ds.Character.name,
    ds.Character.friends.select(ds.Character.name),
)

Once your root query fields are defined, you can put them in an operation using DSLQuery, DSLMutation or DSLSubscription:

DSLQuery(
    ds.Query.hero.select(
        ds.Character.id,
        ds.Character.name,
        ds.Character.friends.select(ds.Character.name),
    )
)

Once your operations are defined, use the dsl_gql function to convert your operations into a document which will be able to get executed in the client or a session:

query = dsl_gql(
    DSLQuery(
        ds.Query.hero.select(
            ds.Character.id,
            ds.Character.name,
            ds.Character.friends.select(ds.Character.name),
        )
    )
)

result = client.execute(query)

Arguments

It is possible to add arguments to any field simply by calling it with the required arguments:

ds.Query.human(id="1000").select(ds.Human.name)

It can also be done using the args method:

ds.Query.human.args(id="1000").select(ds.Human.name)

Aliases

You can set an alias of a field using the alias method:

ds.Query.human.args(id=1000).alias("luke").select(ds.Character.name)

It is also possible to set the alias directly using keyword arguments of an operation:

DSLQuery(
    luke=ds.Query.human.args(id=1000).select(ds.Character.name)
)

Or using keyword arguments in the select method:

ds.Query.hero.select(
    my_name=ds.Character.name
)

Mutations

For the mutations, you need to start from root fields starting from ds.Mutation then you need to create the GraphQL operation using the class DSLMutation. Example:

query = dsl_gql(
    DSLMutation(
        ds.Mutation.createReview.args(
            episode=6, review={"stars": 5, "commentary": "This is a great movie!"}
        ).select(ds.Review.stars, ds.Review.commentary)
    )
)

Variable arguments

To provide variables instead of argument values directly for an operation, you have to:

  • Instantiate a DSLVariableDefinitions:

    var = DSLVariableDefinitions()
    
  • From this instance you can generate DSLVariable instances and provide them as the value of the arguments:

    ds.Mutation.createReview.args(review=var.review, episode=var.episode)
    
  • Once the operation has been defined, you have to save the variable definitions used in it:

    operation.variable_definitions = var
    

The following code:

var = DSLVariableDefinitions()
op = DSLMutation(
    ds.Mutation.createReview.args(review=var.review, episode=var.episode).select(
        ds.Review.stars, ds.Review.commentary
    )
)
op.variable_definitions = var
query = dsl_gql(op)

will generate a query equivalent to:

mutation ($review: ReviewInput, $episode: Episode) {
  createReview(review: $review, episode: $episode) {
    stars
    commentary
  }
}

Subscriptions

For the subscriptions, you need to start from root fields starting from ds.Subscription then you need to create the GraphQL operation using the class DSLSubscription. Example:

query = dsl_gql(
    DSLSubscription(
        ds.Subscription.reviewAdded(episode=6).select(ds.Review.stars, ds.Review.commentary)
    )
)

Multiple fields in an operation

It is possible to create an operation with multiple fields:

DSLQuery(
    ds.Query.hero.select(ds.Character.name),
    hero_of_episode_5=ds.Query.hero(episode=5).select(ds.Character.name),
)

Operation name

You can set the operation name of an operation using a keyword argument to dsl_gql:

query = dsl_gql(
    GetHeroName=DSLQuery(ds.Query.hero.select(ds.Character.name))
)

will generate the request:

query GetHeroName {
    hero {
        name
    }
}

Multiple operations in a document

It is possible to create an Document with multiple operations:

query = dsl_gql(
    operation_name_1=DSLQuery( ... ),
    operation_name_2=DSLQuery( ... ),
    operation_name_3=DSLMutation( ... ),
)

Fragments

To define a Fragment, you have to:

  • Instantiate a DSLFragment with a name:

    name_and_appearances = DSLFragment("NameAndAppearances")
    
  • Provide the GraphQL type of the fragment with the on method:

    name_and_appearances.on(ds.Character)
    
  • Add children fields using the select method:

    name_and_appearances.select(ds.Character.name, ds.Character.appearsIn)
    

Once your fragment is defined, to use it you should:

  • select it as a field somewhere in your query:

    query_with_fragment = DSLQuery(ds.Query.hero.select(name_and_appearances))
    
  • add it as an argument of dsl_gql with your query:

    query = dsl_gql(name_and_appearances, query_with_fragment)
    

The above example will generate the following request:

fragment NameAndAppearances on Character {
    name
    appearsIn
}

{
    hero {
        ...NameAndAppearances
    }
}

Inline Fragments

To define an Inline Fragment, you have to:

  • Instantiate a DSLInlineFragment:

    human_fragment = DSLInlineFragment()
    
  • Provide the GraphQL type of the fragment with the on method:

    human_fragment.on(ds.Human)
    
  • Add children fields using the select method:

    human_fragment.select(ds.Human.homePlanet)
    

Once your inline fragment is defined, to use it you should:

  • select it as a field somewhere in your query:

    query_with_inline_fragment = ds.Query.hero.args(episode=6).select(
        ds.Character.name,
        human_fragment
    )
    

The above example will generate the following request:

hero(episode: JEDI) {
    name
    ... on Human {
      homePlanet
    }
}

Note: because the on and select methods return self, this can be written in a concise manner:

query_with_inline_fragment = ds.Query.hero.args(episode=6).select(
    ds.Character.name,
    DSLInlineFragment().on(ds.Human).select(ds.Human.homePlanet)
)

Meta-fields

To define meta-fields (__typename, __schema and __type), you can use the DSLMetaField class:

query = ds.Query.hero.select(
    ds.Character.name,
    DSLMetaField("__typename")
)

Executable examples

Async example

import asyncio

from gql import Client
from gql.dsl import DSLQuery, DSLSchema, dsl_gql
from gql.transport.aiohttp import AIOHTTPTransport


async def main():

    transport = AIOHTTPTransport(url="https://countries.trevorblades.com/graphql")

    client = Client(transport=transport, fetch_schema_from_transport=True)

    # Using `async with` on the client will start a connection on the transport
    # and provide a `session` variable to execute queries on this connection.
    # Because we requested to fetch the schema from the transport,
    # GQL will fetch the schema just after the establishment of the first session
    async with client as session:

        # Instantiate the root of the DSL Schema as ds
        ds = DSLSchema(client.schema)

        # Create the query using dynamically generated attributes from ds
        query = dsl_gql(
            DSLQuery(
                ds.Query.continents(filter={"code": {"eq": "EU"}}).select(
                    ds.Continent.code, ds.Continent.name
                )
            )
        )

        result = await session.execute(query)
        print(result)

        # This can also be written as:

        # I want to query the continents
        query_continents = ds.Query.continents

        # I want to get only the continents with code equal to "EU"
        query_continents(filter={"code": {"eq": "EU"}})

        # I want this query to return the code and name fields
        query_continents.select(ds.Continent.code)
        query_continents.select(ds.Continent.name)

        # I generate a document from my query to be able to execute it
        query = dsl_gql(DSLQuery(query_continents))

        # Execute the query
        result = await session.execute(query)
        print(result)


asyncio.run(main())

Sync example

from gql import Client
from gql.dsl import DSLQuery, DSLSchema, dsl_gql
from gql.transport.requests import RequestsHTTPTransport

transport = RequestsHTTPTransport(
    url="https://countries.trevorblades.com/", verify=True, retries=3,
)

client = Client(transport=transport, fetch_schema_from_transport=True)

# Using `with` on the sync client will start a connection on the transport
# and provide a `session` variable to execute queries on this connection.
# Because we requested to fetch the schema from the transport,
# GQL will fetch the schema just after the establishment of the first session
with client as session:

    # We should have received the schema now that the session is established
    assert client.schema is not None

    # Instantiate the root of the DSL Schema as ds
    ds = DSLSchema(client.schema)

    # Create the query using dynamically generated attributes from ds
    query = dsl_gql(
        DSLQuery(ds.Query.continents.select(ds.Continent.code, ds.Continent.name))
    )

    result = session.execute(query)
    print(result)