Graphene 3 Queries, Mutations & Subscriptions

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]
export APP_KEY="*******************"
export APP_KEY_SECRET="********************************"
export ACCESS_TOKEN="***************************************"
export ACCESS_TOKEN_SECRET="**********************************"

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!

Query

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

The Mutation

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

Subscription

The subscription is remarkably straightforward.

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]).

Server

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

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
}
}
}
mutation {
updateStatus(status: "I'm posting to twitter from graphene v3") {
createdAt
text
user {
name
screenName
}
}
}
subscription {
filter (track: ["python"]) {
text
user {
name
screenName
}
}
}

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store