GraphQL in Python

Query and mutate data on any GraphQL server from Python services.

Python services query and mutate data on any system that exposes a GraphQL API - Microsoft 365, GitHub, Shopify, Hasura or a custom backend - through outgoing connections. The connection handles transport, timeouts and authentication.

You can write queries in two ways:

String queries: pass a query or mutation as a string to .execute() - use them when the query is known up front
DSL queries: build the query from Python objects with the gql DSL - use them when the query's structure depends on runtime conditions

Create a connection

To create a connection, go to Connections > Outgoing > GraphQL in the Dashboard, click Create a new connection and fill in the form:

Name: Airport GraphQL
Address: your GraphQL server's address, e.g. https://graphql.example.com
Click OK

The connection is ready the moment you click OK and every service can use it through self.out.graphql, passing the connection's name.

Query a server

To run a query, pass it as a string to .execute() - the connection parses the string before sending it to the server. The service below queries flights that are currently airborne:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class GetActiveFlights(Service):

    def handle(self):

        # Get the connection by the name configured in Dashboard
        conn = self.out.graphql['Airport GraphQL']

        # Query all flights that are currently airborne
        query = """
            {
                flights(status: "airborne") {
                    id
                    flight_number
                    origin
                    destination
                    status
                }
            }
        """

        result = conn.execute(query)

        self.logger.info('Flights: %s', result)

.invoke() is an alias for .execute() - both methods work the same way.

Query with variables

To provide GraphQL variables, pass a params dict to .execute():

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class GetFlightById(Service):

    input = 'flight_id'

    def handle(self):
        conn = self.out.graphql['Airport GraphQL']

        query = """
            query GetFlight($flight_id: ID!) {
                flight(id: $flight_id) {
                    id
                    flight_number
                    origin
                    destination
                    departure_time
                    gate {
                        name
                        terminal
                    }
                }
            }
        """

        params = {'flight_id': self.request.input.flight_id}
        result = conn.execute(query, params=params)

        self.response.payload = result

Run mutations

Mutations work the same way as queries - call .execute() with a mutation string and optional variables:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class AssignFlightGate(Service):

    input = 'flight_id', 'gate_id'

    def handle(self):
        conn = self.out.graphql['Airport GraphQL']

        mutation = """
            mutation AssignGate($flight_id: ID!, $gate_id: ID!) {
                assignGate(flightId: $flight_id, gateId: $gate_id) {
                    id
                    flight_number
                    gate {
                        name
                        terminal
                    }
                }
            }
        """

        params = {
            'flight_id': self.request.input.flight_id,
            'gate_id': self.request.input.gate_id,
        }

        result = conn.execute(mutation, params=params)

        self.logger.info('Gate assigned: %s', result)

Build queries with the DSL

The gql library's DSL module lets you build queries from Python objects instead of strings, with auto-completion in your IDE and field validation against the server's schema.

To use it, call .session() to open a connection that fetches the schema, then build the query with the DSLSchema object:

# -*- coding: utf-8 -*-

# gql
from gql.dsl import DSLQuery, dsl_gql

# Zato
from zato.server.service import Service

class GetFlightsDSL(Service):

    def handle(self):
        conn = self.out.graphql['Airport GraphQL']

        with conn.session() as (session, schema):

            query = dsl_gql(
                DSLQuery(
                    schema.Query.flights.select(
                        schema.Flight.id,
                        schema.Flight.flight_number,
                        schema.Flight.origin,
                        schema.Flight.destination,
                    )
                )
            )

            result = session.execute(query)

        self.logger.info('Flights: %s', result)

Use the DSL when the query's structure depends on runtime conditions:

# -*- coding: utf-8 -*-

# gql
from gql.dsl import DSLQuery, dsl_gql

# Zato
from zato.server.service import Service

class GetFlightsConditional(Service):

    def handle(self):
        conn = self.out.graphql['Airport GraphQL']

        with conn.session() as (session, schema):

            # Start with core fields
            fields = [schema.Flight.id, schema.Flight.flight_number, schema.Flight.status]

            # Add optional fields based on the caller's request
            if self.request.input.include_gate:
                fields.append(schema.Flight.gate)

            if self.request.input.include_crew:
                fields.append(schema.Flight.crew)

            query = dsl_gql(
                DSLQuery(
                    schema.Query.flights.select(*fields)
                )
            )

            result = session.execute(query)

        self.logger.info('Flights: %s', result)

Ping the server

To verify that the connection works, call .ping() - it sends a schema introspection query and returns True when the server responds:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class PingGraphQL(Service):

    def handle(self):
        conn = self.out.graphql['Airport GraphQL']

        is_alive = conn.ping()
        self.logger.info('Server alive: %s', is_alive)

Security

GraphQL connections are secured the same way as all other outgoing connections - by attaching a security definition. The definition is created separately in Security in the Dashboard and assigned to the connection when you create or edit it.

The following security types are supported:

Basic Auth - sends a username and password as HTTP Basic Authentication headers with every request to the GraphQL server
API key - sends a token in a configurable HTTP header, commonly used with services like GitHub's GraphQL API where the header is Authorization: Bearer <token>
OAuth - obtains a Bearer token from an OAuth endpoint and attaches it to each request, commonly used with Microsoft 365 and other enterprise GraphQL APIs

The connection attaches the credentials to every request - your service code stays the same regardless of the security mechanism.

Security definitions can also be managed through enmasse for automated deployments.

Custom headers

To send additional HTTP headers with every request to the GraphQL server, use the Extra headers field on the connection. It accepts a JSON object where each key is a header name and each value is the header value.

For example, to send a tenant identifier and a custom tracing header:

{"X-Tenant": "acme", "X-Request-Source": "zato"}

The connection merges these headers into every request, whether made with .execute() or .session().

Feature	What it does
Shopify GraphQL Admin API	A production GraphQL API used from Python services
REST outgoing connections	Call external REST APIs when a system offers no GraphQL
gRPC	Call gRPC endpoints with Protocol Buffers
Enmasse	Define connections and security in YAML for automated deployments