Last updated: by Valentino Gagliardi

GraphQL subscriptions in Django with Ariadne and Channels

... and what it takes to work with the Django ORM in an asynchronous context.

Adding GraphQL subscriptions to Django with Ariadne and Channels

WebSockets are mostly associated with the asynchronous capabilities of JavaScript engines. But it doesn't have to be JavaScript all the way down.

The Python asynchronous ecosystem has become robust and stable enough in recent years to offer a solid ground for real-time projects.

In this post we will see how to add GraphQL subscriptions to Django with Ariadne and Channels, and what it takes to work with the Django ORM in an asynchronous context.

What is Ariadne?

Ariadne is a GraphQL library for Python, which offers also an integration for Django.

GraphQL is a data query language which allows the client to precisely define what data to fetch from the server and combine data from multiple resources in one request. In a sense, this is what we always did with REST APIs, but GraphQL takes this a step further, pushing more control to the client.

What is Django Channels?

Channels is a library which adds, among other protocols, WebSockets capabilities to Django.

Setup and sample models

This mini tutorial assumes you have a Django project where you can install Django Channels and Ariadne.

My demo project has a sample model named Order, as follows:

from django.db import models
from django.conf import settings


class Order(models.Model):
    class State(models.TextChoices):
        PAID = "PAID"
        UNPAID = "UNPAID"

    date = models.DateField()
    state = models.CharField(max_length=6, choices=State.choices, default=State.UNPAID)
    user = models.ForeignKey(to=settings.AUTH_USER_MODEL, on_delete=models.CASCADE)

To start off, with the virtual Python environment active, install Channels and Ariadne:

pip install ariadne channels

Next up, configure INSTALLED_APPS to load both apps, and ASGI_APPLICATION:

INSTALLED_APPS = [
    ...
    "ariadne.contrib.django",
    "channels",
]

ASGI_APPLICATION = "django_project.asgi.application"

Once this is done, open up asgi.py, which in a stock Django project is located in django_project/asgi.py (where django_project is my sample project), and configure the ASGI application so that it responds with URLRouter from Channels:

import os

from django.core.asgi import get_asgi_application

from ariadne.asgi import GraphQL
from channels.routing import URLRouter
from django.urls import path

from orders.schema import schema

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "django_project.settings")


application = URLRouter(
    [
        path("graphql/", GraphQL(schema)),
        path("", get_asgi_application())
    ]
)

In the URLRouter configuration we load two routes. First:

path("graphql/", GraphQL(schema))

which is the Ariadne ASGI application. It handles both HTTP, for GraphQL queries, and WebSockets, once the connection is established after a successful subscription request.

Next up there is:

path("", get_asgi_application())

which handles any other request to Django, for example /admin, or other non-GraphQL routes.

Note: Ariadne documentation suggests to use AsgiHandler from channels.http instead of get_asgi_application(), but AsgiHandler has been deprecated in favor of the latter.

Schema and imports

With the configuration and the URLs in place, we can set up the schema, and the Ariadne machinery.

For the scope of this post we build the schema in the app folder at orders/schema.py. The following code shows how to declare the schema in Ariadne, and what modules we need to import:

import asyncio

from ariadne import gql, make_executable_schema
from ariadne import SubscriptionType
from ariadne.contrib.django.scalars import date_scalar

from orders.models import Order

type_defs = gql(
    """
        scalar Date
        
        enum OrderState {
            PAID
            UNPAID
        }
        
        type User {
            username: String
            email: String
        }
        
        type Order {
            date: Date
            state: OrderState
            user: User
        }
        
        type Query {
            getOrders: [Order]
        }
        
        type Subscription {
            getOrder: Order
        }
"""
)

As you can see from the schema, we define a root Query type, which is mandatory, but we don't declare a resolver to handle it. This is intended for the purpose of this tutorial since we want to focus only on subscriptions.

In fact, we can see a Subscription root type, which for now declares just one field:

type Subscription {
    getOrder: Order
}

This declaration makes possible to handle the following query sent from the client:

subscription getLastOrder {
  getOrder {
    state
    date
  }
}

With the schema in place, we can move to define a resolver in order to satisfy the subscription.

Subscriptions, generators, and resolvers

To satisfy a subscription in Ariadne, we need to define two functions:

  • a asynchronous generator, in charge of fetching the actual data from the database
  • a resolver, in charge of returning data to the client

A generator function in Python is a function which can be paused and resumed at will. The concept is not so different from JavaScript generator functions.

Generator functions return simple values to the caller, and work synchronously. Asynchronous generator function instead work asynchronously, and return an asynchronous generator object.

This kind of objects can be iterated over with async for in Python. Again, the concept is similar to asynchronous generator functions in JavaScript, and if you're a JavaScript developer the theory overlaps perfectly.

What matter for us here, is that Ariadne wants an asynchronous generator function to fetch data from the database, and another resolver function which basically sends back data to the client.

The following example shows how to wire up the two functions:

"""
Your imports and the schema here ...
"""

subscription = SubscriptionType()


@subscription.source("getOrder")
async def generate_order(obj, info):
    """Disclaimer: for demo purpose only."""
    while True:
        await asyncio.sleep(1)
        # TODO: get the data from the database


@subscription.field("getOrder")
def resolve_order(order, info):
    return order


schema = make_executable_schema(type_defs, [date_scalar], subscription)

Here we have an asynchronous generator decorated with @subscription.source(), and a resolver decorated with @subscription.field().

On the last line, we wire up the schema, the Date scalar, and the subscription:

schema = make_executable_schema(type_defs, [date_scalar], subscription)

This schema is now ready to be picked up by django_project/asgi.py.

Now that we have the big picture, let's focus on the interaction between the asynchronous world and the Django ORM.

Working asynchronously with the Django ORM

Since the arrival of asynchronous views in Django, it has been stressed over and over that while Django is gaining asynchronous capabilities, the ORM is still synchronous.

What this means is that the following code is wrong:

@subscription.source("getOrder")
async def generate_order(obj, info):
    """Disclaimer: for demo purpose only."""
    while True:
        await asyncio.sleep(1)
        """Wrong!"""
        yield Order.objects.last()

If we try to access the ORM from an asynchronous context, Django screams to us:

"You cannot call this from an async context - use a thread or sync_to_async."

You can test this out by heading over http://127.0.0.1:8000/graphql/ with the Django project running in development, and by sending out a subscription request:

You cannot call this from an async context - use a thread or sync_to_async.

To make the ORM work in an asynchronous context, in the latest versions of Django we can peruse sync_to_async from asgiref.sync.

Here, since we work with Django Channels, we need to use database_sync_to_async, a small wrapper around asgiref.sync.sync_to_async.

The following example shows how to interact with the ORM from an asynchronous context:

...
from channels.db import database_sync_to_async
...

...
@subscription.source("getOrder")
async def generate_order(obj, info):
    """Disclaimer: for demo purpose only."""
    while True:
        await asyncio.sleep(1)
        yield await database_sync_to_async(lambda: Order.objects.last())()
...

This change will make the subscription work:

The GraphQL subscriptions is working correctly in Django with Ariadne and Channels

We made Django work with Ariadne and Channels for a simple subscription:

subscription getLastOrder {
  getOrder {
    state
    date
  }
}

However, things start to become tricky if we try to reach for the foreign key relationship, like in the following subscription:

subscription getLastOrder {
  getOrder {
    state
    date
    user {
      username
      email
    }
  }
}

By launching this subscription, the Django ORM will raise the same exception as in our first experiment:

"You cannot call this from an async context - use a thread or sync_to_async."

This time, the error comes from the user field access. What can we do about it?

To force the foreign key query evaluation in the asynchronous context, we can use .select_related() from the Django ORM, as in the following example:

@subscription.source("getOrder")
async def generate_order(obj, info):
    """Disclaimer: for demo purpose only."""
    while True:
        await asyncio.sleep(1)
        yield await database_sync_to_async(
            lambda: Order.objects.select_related("user").last()
        )()

With Order.objects.select_related("user").last() we issue just one slightly bigger query, forcing Django to fetch both the Order, and the related user in the asynchronous context.

Subscriptions in the frontend, and CORS

To consume GraphQL subscriptions in the frontend, Apollo client is the way to go. The documentation has a complete example of how to make this happen.

Basically, you will need HttpLink and WebSocketLink, and for React, you can use useSubscription().

As for the backend, if the frontend lives on a different origin than the GraphQL API, you will need to enable CORS.

Out of the box, Ariadne offers no CORS support, but this is easily solvable with Starlette's CORSMiddleware as shown in the following example from django_project/asgi.py:

...
from starlette.middleware.cors import CORSMiddleware
...


application = URLRouter(
    [
        path(
            "graphql/",
            CORSMiddleware(
                GraphQL(schema),
                allow_origins=["https://first-origin.io/"],
                allow_methods=["*"]
            ),
        ),
        path("", get_asgi_application()),
    ]
)

Hope you learned something new today. Thanks for reading!

Further resources

Valentino Gagliardi

Hi! I'm Valentino! I'm a freelance consultant with a wealth of experience in the IT industry. I spent the last years as a frontend consultant, providing advice and help, coaching and training on JavaScript, testing, and software development. Let's get in touch!

More from the blog: