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.


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 have a love hate relationship with the sphinx document generator. It is a hugely powerful set of tools, but I find the learning curve frustratingly steep.

What I really want to do is use markdown. The mkdocs project uses markdown as it’s document syntax, so I’m ready to go! but wait; how can I automatically generate the documentation from docstrings in my code?

After a little googling I discovered an excellent package called mkautodoc which is used by the encode group. This was close to what I was looking for, but it uses a custom docstring format, and I…

Maintaining valid types when deserializing JSON with Python is hard. This article discusses an approach using typing. You can find the package on GitHub.

The Problem

Here we have a python dict with a mix of data types.

from datetime import datetime
from decimal import Decimal

dct = {
'some_text': 'Hello, World!',
'some_date': datetime.fromisoformat('2020-01-29T12:56:13'),
'some_int': 42,
'some_float': 3.14,
'some_decimal': Decimal("2.414"),
'some_list': [
'other_text': 'Hello, World!',
'other_date': datetime.fromisoformat('2020-01-29T12:56:13'),
'other_decimal': Decimal("2.414")
'other_text': 'Hello, World!',
'other_date': datetime.fromisoformat('2020-01-29T12:56:13'),
'other_decimal': Decimal('2.414')

Because the dict contains datetime and Decimal this won’t serialize without some extra work:

import json try: text = json.dumps(dct)…

The next generation of Python web servers will be powered by ASGI, which is a low level standard for asynchronous web servers. A number of frameworks have been written to support this standard. I will talk about how to get started with bareASGI, a framework I have written and have been using for about 6 months.

ASGI Servers

To use ASGI you need a server. There are several to choose from. Here we will use hypercorn.

$ pip install hypercorn

Hello, World!

Let’s get started!

import asyncio
from hypercorn.asyncio import serve
from hypercorn.config import Config
from bareasgi import Application, bytes_writer
app = Application()@app.on_http_request({'GET'}…

I’ve been using EventSource for some time now for streaming GraphQL subscriptions, and it’s fantastic, but there are two reasons why it sucks.

  1. It uses an arcane text based protocol,
  2. The only way to communicate the initial request is with the url.

The second has started to cause me problems. When my GraphQL subscription queries become to large I get unexpected errors which are hard to debug.

Under the hood we know that the EventSource is simply a streaming fetch using the GET method. If only there were a way to do a streaming fetch as a POST we could…

While writing some server side code I needed to block until all tasks for a particular component had completed; like a semaphore in reverse. After some googling failed to provide a ready made solution I rolled my own.

from asyncio import Eventclass CountdownEvent:    def __init__(self)-> None:
self._event = Event()
self._count = 0
def increment(self) -> int:
self._count += 1
return self._count
def decrement(self) -> int:
assert self._count > 0, "Count cannot go below zero"
self._count -= 1
if self._count == 0:
return self._count
async def wait(self) -> None:
await self._event.wait()
@property def count(self) -> int…

I’ve shed many tears with the following exception:

got Future <Future pending> attached to a different loop

Here’s a simple example which sets and clears an asyncio.Event(), via a web page served by the Uvicorn ASGI server, using the bareASGI framework.

import asyncio
import urllib.parse
from bareasgi import (
import uvicorn
async def get_index(scope, info, matches, content):
"""Change an event"""
# !!! Do something asynchronous !!!
await asyncio.wait_for(info['event'].wait(), timeout=0.1)
except asyncio.TimeoutError:
text = """ <html> <body> <button onclick="changeEvent('{state}')">{message}</button> <script> function changeEvent(state) {{ fetch('/change_event', {{ method: 'POST', body: state }}) .then(response => location.reload()) }} </script>…

My use case was to handle the graceful termination of a RabbitMQ consumer written in Python 3.7 using asycio (with aio_pika). I didn’t want the task to be simply interrupted, since if it was in the middle of handling a new message from the queue I wouldn’t know if the item had been successfully processed.

The pattern I’m using is to wait on both a the task I’m interested in and an asyncio.Event object. This looks as follows:

import asyncio

cancellation_event = asyncio.Event()
done, pending = await asyncio.wait(
[cancellation_event.wait(), asyncio.sleep(30)],

If the cancellation event is set before the…

There are man articles on CloudFlare’s PKI toolkit, but no single article had all the tasks detailed for the pattern I use, so I thought I’d put them here.

Installation on Linux

Unfortunately, at the time of writing, the latest packaged version (1.2) contains a bug that makes it impossible to create certificates with hosts, so the software must be installed with Go.

$ sudo apt install golang
$ go get -u
$ sudo cp ~/go/bin/cfssl /usr/local/bin/cfssl
$ go get -u
$ sudo cp ~/go/bin/cfssljson /usr/local/bin/cfssljson

The Pattern

The root of all the certificates is a certificate authority (or “CA”) from which all…

I’ve found that docker-compose provides a clean environment for developing applications, and I wanted to try out change streams to create a horizontally scaleable Python app.

A quick search provided a selection of docker-compose files. All of them created a cluster, but when I tried to connect, Python hung! This is what I got working.

The Python package that provides async access to MongoDB is motor. The documentation for creating the connection refers me to the underlying PyMongo documentation for connecting to a cluster:

MongoClient('localhost', replicaset='foo')

Turning to the MongoDB replica set configuration, we can see that the cluster…

Rob Blackbourn

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