Graphene 3 Queries, Mutations & Subscriptions

4 min readOct 8, 2020

The next major release of Graphene (version 3) introduces support for asyncio methods. In this post I’ll write a small GraphQL service using Twitter as the source of data. You can find the source code on GitHub.

Prerequisites

If you want to run the code you’ll need a Twitter developer account, with a “project” that has api keys and access tokens. I ran this code on a Linux box running Ubuntu 20.04 LTS with Python 3.8. There’s a fix in graphene regarding subscriptions that has not been released at the time of writing (8 Oct 2020). I’ve created a patched version (pip install jetblack-graphene).

Twitter API

I’m using jetblack-tweeter. This supports pluggable HTTP clients, and the client must be specified when installing the package. I’m using bareClient as it works well with the ASGI server I’ll be using.

pip install jetblack-tweeter[bareclient]

We’ll follow best practice and set up some environment variables to hold the Twitter codes and secrets rather than saving them in our code.

export APP_KEY="*******************"
export APP_KEY_SECRET="********************************"
export ACCESS_TOKEN="***************************************"
export ACCESS_TOKEN_SECRET="**********************************"

Now we can try the three API calls we will be using. The query we will search tweets, the mutation will update the status, and the subscription will filter incoming tweets.

If all went well you should see a print out of search results, have updated your twitter status, and be seeing updates as tweets are received with the hashtag “#python”.

Types

The search function hits the endpoint https://api.twitter.com/1.1/search/tweets.json which is documented here. It returns a bunch of JSON which we will need to map to graphene objects in python. Here is what the code looks like for the types. There’s a lot of code, as Twitter returns a lot of data!

There are a few points to note here. I didn’t have to map all the data (in fact I’ve left a few things out). Any unmapped data will just be discarded. Some fields have been marked required=True to add extra validation, and indicate to the consumer things that will always exist. I’ve used the BigInt data type, as I notice twitter ids are huge (I could have used the ID type instead). Lastly, the data is nested, and we can see this implemented by wrapping the nested type in a graphene.Field.

Query

The query will take two arguments, a mandatary search string and an optional count to limit the number of tweets returned.

You can see how the query is named search_tweets and the function to resolve it is named resolve_search_tweets. All the names get camel-cased, so the consumer will see this as searchTweets. The input arguments are declared on lines 11–12 and consumed on lines 19–20.

Note that we don’t need to return the graphene type objects we declared above. As long as the dictionary has the same shape and fields as the types, graphene will do the rest for us.

There’s a bit of magic on line 22. We’ll see later on how to pass the Twitter client through to the resolver.

The Mutation

For the mutation we will take a single argument, the status, and return the generated tweet.

Note how mutations are created as a standalone classes which inherit from graphene.Mutation, and then composed into a single class which inherits from graphene.ObjectType. The mutation itself has a single method mutate to handle the mutation. Line 12 declares the output type of the mutation.

Subscription

The subscription is remarkably straightforward.

You can see how we use async for and yield to stream the tweets to the client.

Schema

The queries, mutations and subscriptions can now be assembled into a schema.

Application

I’m going to use an ASGI web application framework to handle the GraphQL protocol. I’m using bareASGI-graphql-next. You will need to specify the optional graphene package when installing (pip install bareasgi-graphql-next[graphene]).

You can see on line 28 how we pass the Twitter client through to the resolvers using the shared info attribute. Graphene support is added on line 32 using the schema we created above.

Server

The last thing to do is to serve our ASGI application. I’m using Hypercorn.

If you run this a web service will be served from http://0.0.0.0:10001.

Altair

You can test the service using the Chrome extension Altair GraphQL client. Enter the address for the query endpoint (http://0.0.0.0:10001/graphql) and run the following query. Note that the count argument is optional.

query {
  searchTweets(q: "python", count: 5) {
    statuses {
      createdAt
      text
      user {
        name
        screenName
      }
    }
    searchMetadata {
      completedIn
      query
      count
    }
  } 
}

Because we’re using GraphQL we can choose which fields we get back. Nice!

Here is an example of a mutation which posts a tweet.

mutation {
  updateStatus(status: "I'm posting to twitter from graphene v3") {
    createdAt
    text
    user {
      name
      screenName
    }
  } 
}

Now let’s try the subscription. When we try to run it a dialog should pop up asking for the subscription endpoint (ws://0.0.0.0:10001/subscriptions).

subscription {
  filter (track: ["python"]) {
    text
    user {
      name
      screenName
    }
  } 
}

Hopefully you will now have a ticking report with tweets containing the word “python”.