Getting started with GraphQL in Python with FastAPI and Ariadne

Yasser Tahiri
October 5th, 2021 · 5 min read

FastAPI is a high-performance framework for building web APIs with Python. Its simple and intuitive nature makes it easy to quickly develop robust web APIs using very little boilerplate code. In this article, we’ll introduce FastAPI and how to set up a GraphQL server with it using Graphene & Ariadne.

From the official docs, building web applications with FastAPI reduce about 40 percent of developer-induced errors, and this is made possible through the use of Python 3.6 type declarations. With all its features, including the automatic generation of interactive API documentation, building web apps with Python has never been easier.

Setting up our app

Before we get started, let’s confirm that we have Python 3.7+ installed by running the following command on our terminal:

1python --version

Note: If you don’t have it installed, get it here.

  • To build this project we will need the following Python packages:

    • FastAPI - A fast, modern, and flexible framework for building web APIs with Python.
    • Ariadne - A Python library for implementing GraphQL servers using schema-first approach.

Let’s go ahead and install these packages in our virtual environment:


We install uvicorn with the “standard” option since that brings optional extras like WebSocket support.

Asynchronous servers in Python

ASGI is an emerging standard for building asynchronous services in Python that support HTTP/2 and WebSocket. Web frameworks like Flask and Pyramid are examples of WSGI based frameworks and do not support ASGI. Django was for a long time a WSGI based framework but it has introduced ASGI support in version 3.1.

  • ASGI has two components:

    • Protocol Server: Handles low level details of sockets, translating them to connections and relaying them to the application
    • Application: A callable that is responsible for handling incoming requests. There are several ASGI frameworks that simplify building applications.

As an application developer, you might find that you will be mostly working at the application and framework levels.

Examples of ASGI servers include Uvicorn, Daphne and Hypercorn. Examples of ASGI frameworks include Starlette, Django channels, FastAPI and Quart.

Ariadne provides a GraphQL class that implements an ASGI application so we will not need an ASGI framework. We will use the uvicorn server to run our application.

Note: Understand the difference between ASGI and WSGI Blog by Raoof Naushad.

Async in Python using asyncio

The asyncio library adds async support to Python. To declare a function as asynchronous, we add the keyword async before the function definition as follows:

1async def hello_world():
2 return "Hello world"
  • Using await, we would call our asynchronous function as follows:
1obytes = await hello_world()

Writing the GraphQL schema

  • Create a file called schema.graphql. We will use it to define our GraphQL schema.

Custom types:

Our schema will include five custom types, described below.

1type User {
2 id: ID! // This is the id of the user
3 email: String! // This is the email of the user
4 password: String! // This is the password of the user
7type blog {
8 id: ID! // This is the id of the blog
9 title: String! // This is the title of the blog
10 description: String! // This is the description of the blog
11 completed: Boolean! // This is the completed status of the blog
12 ownerId: ID! // This is the id of the owner of the blog
15type blogResult {
16 errors: [String] // This is the list of errors
17 blog: blog // This is the blog
20type blogsResult {
21 errors: [String]
22 blogs: [blog] // This is the list of blogs
25type InsertResult {
26 errors: [String]
27 id: ID // This is the id of the inserted blog
30type TokenResult {
31 errors: [String]
32 token: String // This is the token
  • After Define the schema, lets add the Query, Mutation, Subscription and Type definitions.
1schema {
2 query: Query // This is the query type
3 mutation: Mutation // This is the mutation type
4 subscription: Subscription // This is the subscription type
7type Query {
8 blogs: blogsResult! // This is the list of blogs
9 blog(blogId: ID!): blogResult! // This is the blog
12type Mutation {
13 createblog(title: String!, description: String!): InsertResult! // This is the blog
14 createUser(email: String!, password: String!): InsertResult! // This is the user
15 createToken(email: String!, password: String!): TokenResult! // This is the token
18type Subscription {
19 reviewblog(token:String!): InsertResult! // This is the blog

Setting up the project

Create a file called and add the code below:

1from ariadne import QueryType, make_executable_schema, load_schema_from_path
2from ariadne.asgi import GraphQL
4type_defs = load_schema_from_path("schema.graphql")
6query = QueryType()
10def resolve_hello(*_):
11 return "Hello world!"
14schema = make_executable_schema(type_defs, query)
15app = GraphQL(schema, debug=True)

We read the schema defined in the schema.graphql file and added a simple query called hello that we will use to test that our server is running. Our server is now ready to accept requests.

  • Start the server by running:
1uvicorn app:app --reload
  • Open the GraphQL PlayGround by visiting http://localhost:8000. Paste the hello query below and hit the “Play” button:
1query {
2 hello

Congratulations, your GraphQL server is running 🥳!

Once you confirm that the server is running fine, you can delete the resolve_hello function from and delete the hello query in the type Query section of schema.graphql.

Storing users and blogs

Since this article discusses GraphQL operations with an emphasis on subscriptions, we will skip the database component entirely and store our data in memory. We will use two variables for this:

  • users: A python dictionary where the keys are usernames and the values are the user details.
  • blogs: A python list which will store all blogs

Create a file called Initialize users, blogs & queues to an empty list.

1users = {}
2blog = []
3queues = []

Define API & GraphQL


Defining the mutations

Let’s add resolvers for the mutations defined in the schema. These will live inside a file called Go ahead and create it.

First add the createUser resolver to

1from ariadne import ObjectType, convert_kwargs_to_snake_case
3from store import users, blogs
5mutation = ObjectType("Mutation")
10async def resolve_create_user(obj, info, email, password):
11 try:
12 if not users.get(username):
13 user = {
14 "id": len(users) + 1,
15 "email": email,
16 "password": password,
17 }
18 users[username] = user
19 return {
20 "success": True,
21 "user": user
22 }
23 return {
24 "success": False,
25 "errors": ["Username is taken"]
26 }
28 except Exception as error:
29 return {
30 "success": False,
31 "errors": [str(error)]
32 }

We import ObjectType and convert_kwargs_to_snake_case from the Ariadne package. ObjectType is used to define the mutation resolver, and convert_kwargs_to_snake_case recursively converts arguments case from camelCase to snake_case.

We also import users and blogs from, since these are the variables we will use as storage for our users and blogs.

3async def resolve_create_blog(obj, info, content, title, description, completed, ownerId):
4 try:
5 blog = {
6 "ID": id,
7 "title": title,
8 "description": description
9 "completed": completed,
10 "ownerId": ownerId
11 }
12 blogs.append(blog)
13 return {
14 "success": True,
15 "blog": blog
16 }
17 except Exception as error:
18 return {
19 "success": False,
20 "errors": [str(error)]
21 }

In resolve_create_blog, we create a dictionary that stores the attributes of the blog. We append it to the blogs list and return the created blog. If successful, we set success to True and return success and the created blog object. If there was an error, we set success to False and return success and the error blog.

We now have our two resolvers, so we can point Ariadne to them. Make the following changes to

At the top of the file, import the mutations:

1from mutations import mutation

Then add mutation to the list of arguments passed to make_executable_schema:

1schema = make_executable_schema(type_defs, query, mutation)

Big Brain Time

Defining the queries

Now we are ready to implement the two queries of our API. Let’s start with the blogs query. Create a new file, and update it as follows:

  • Create get_blogs resolver.
1# as you know here i use Database[PostgreSQL] to connect to the database
2# Install psycopg2 & databases[postgresql] & asyncpg
4async def get_blogs(
5 skip: Optional[int] = 0, limit: Optional[int] = 100
6) -> Optional[Dict]:
7 query =, limit=limit)
8 result = await database.fetch_all(query=query)
9 return [dict(blog) for blog in result] if result else None
11async def get_blog(blog_id: int) -> Optional[Dict]:
12 query = == int(blog_id))
13 result = await database.fetch_one(query=query)
14 return dict(result) if result else None
  • then we could use it in our schema:
1from typing import Optional
3from ariadne import QueryType, convert_kwargs_to_snake_case
5from crud import get_blogs, get_blog # Create a file called and add the get_blogs function
6from schemas.error import MyGraphQLError
10async def resolve_blogs(obj, info, skip: Optional[int] = 0, limit: Optional[int] = 100):
11 blogs = await get_blogs(skip=skip, limit=limit)
13 return {"blogs": blogs}
17async def resolve_blog(obj, info, blog_id):
18 blog = await get_blog(blog_id=blog_id)
20 if not blog:
21 raise MyGraphQLError(code=404, message=f"blog id {blog_id} not found")
23 return {"success": True, "blog": blog}
26query = QueryType()
27query.set_field("blogs", resolve_blogs)
28query.set_field("blog", resolve_blog)


Subscribing to new blogs

New blogs are now being added to subscription queues, but we do not have any queues yet. All that is remaining is implementing our GraphQL subscription to create a queue and add it to the queues list, read blogs from it and push the appropriate ones to the GraphQL client.

In Ariadne, we need to declare two functions for every subscription defined in the schema.

Subscription source

Create a new file, and define our subscription source in it as follows:

1from ariadne import SubscriptionType, convert_kwargs_to_snake_case
2from graphql.type import GraphQLResolveInfo
4subscription = SubscriptionType()
9async def review_blog_resolver(review_blog, info: GraphQLResolveInfo, token: str):
10 return {"id": review_blog}

Note: create a dynamic Source Review for Blog, by using RabbitMQ.

Rabbit MQ is a messaging broker that allows you to publish and subscribe to messages.

1# This is Example from My Project FastQL
4async def review_blog_source(obj, info: GraphQLResolveInfo, token: str):
5 user = await security.get_current_user_by_auth_header(token)
6 if not user:
7 raise MyGraphQLError(code=401, message="User not authenticated")
9 while True:
10 blog_id = await rabbit.consumeblog()
11 if blog_id:
12 yield blog_id
13 else:
14 return


All the hard work is done! Now comes the easy part; seeing the API in action. Open the GraphQL Playground by visiting http://localhost:8000.

Let’s begin by creating two users, user_one and user_two. Paste the following mutation and hit play.

1mutation {
2 createUser(
4 password:"admin"
5 ) {
6 success
7 user {
8 userId
9 email
10 password
11 }
12 }

Once the first user is created, change the username in the mutation from user_one to user_two and hit play again to create the second user.

Now we have two users who can Blog. Our createBlog mutation expects us to provide senderId and recipientId. If you looked at the responses from the createUser mutations you already know what IDs were assigned to them, but let’s assume we only know the usernames, so we will use the userId query to retrieve the IDs.

1query {
2 userId(
3 // User One
5 password: "admin"
6 )
  • Publish your blog to the second user by using the createBlog mutation.
1mutation {
2 createBlog(
3 senderId: "1",
4 recipientId: "2",
5 title:"Blog number1"
6 description:"This is the first blog"
7 ownerId: "1"
8 ) {
9 success
10 blog {
11 title
12 description
13 ownerId
14 recipientId
15 senderId
16 }
17 }


Congratulations for completing. We learnt about ASGI, how to add subscriptions to a GraphQL server built with Ariadne, and using asyncio.Queue.

If you got any questions, please feel free to ask them in the comments.

This was just a simple API to demonstrate how we can use subscriptions to add real time functionality to a GraphQL API. The API could be improved by adding a database, user authentication, allowing users to attach files in Blog, allowing users to delete Blog and adding user profiles.

If you are wondering how you can incorporate a database to an async API, here are two options:

  • aiosqlite: A friendly, async interface to sqlite databases.

I can’t wait to see what you build!


You could also get inspiration from this Project FastQL.

To be honest, this got quite long. If you are patient enough to read this full and find it interesting then please share it, and Follow me on twitter for the next articles.

More articles from Obytes

GO Serverless! Part 2 - Terraform and AWS Lambda External CI

If you didn't read the previous part , we highly advise you to do that! In the previous part we have discussed the pros and cons of…

September 29th, 2021 · 8 min read

GO Serverless Part 1 Serverless Architecture and Components Integration Patterns

Cloud serverless architectures became more and more popular in the past years, and many enterprises started leveraging it and even migrating…

September 15th, 2021 · 18 min read


Our mission and ambition is to challenge the status quo, by doing things differently we nurture our love for craft and technology allowing us to create the unexpected.