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:

1fastapi[uvicorn]
2ariadne

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
5}
6
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
13}
14
15type blogResult {
16 errors: [String] // This is the list of errors
17 blog: blog // This is the blog
18}
19
20type blogsResult {
21 errors: [String]
22 blogs: [blog] // This is the list of blogs
23}
24
25type InsertResult {
26 errors: [String]
27 id: ID // This is the id of the inserted blog
28}
29
30type TokenResult {
31 errors: [String]
32 token: String // This is the token
33}
  • 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
5}
6
7type Query {
8 blogs: blogsResult! // This is the list of blogs
9 blog(blogId: ID!): blogResult! // This is the blog
10}
11
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
16}
17
18type Subscription {
19 reviewblog(token:String!): InsertResult! // This is the blog
20}

Setting up the project

Create a file called app.py and add the code below:

1from ariadne import QueryType, make_executable_schema, load_schema_from_path
2from ariadne.asgi import GraphQL
3
4type_defs = load_schema_from_path("schema.graphql")
5
6query = QueryType()
7
8
9@query.field("hello")
10def resolve_hello(*_):
11 return "Hello world!"
12
13
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
3}

Congratulations, your GraphQL server is running 🥳!

Once you confirm that the server is running fine, you can delete the resolve_hello function from app.py 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 store.py. Initialize users, blogs & queues to an empty list.

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

Define API & GraphQL

GIF

Defining the mutations

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

First add the createUser resolver to mutations.py.

1from ariadne import ObjectType, convert_kwargs_to_snake_case
2
3from store import users, blogs
4
5mutation = ObjectType("Mutation")
6
7
8@mutation.field("createUser")
9@convert_kwargs_to_snake_case
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 }
27
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 store.py, since these are the variables we will use as storage for our users and blogs.

1@mutation.field("createblog")
2@convert_kwargs_to_snake_case
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 app.py:

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, queries.py 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
3
4async def get_blogs(
5 skip: Optional[int] = 0, limit: Optional[int] = 100
6) -> Optional[Dict]:
7 query = blog.select(offset=skip, limit=limit)
8 result = await database.fetch_all(query=query)
9 return [dict(blog) for blog in result] if result else None
10
11async def get_blog(blog_id: int) -> Optional[Dict]:
12 query = blog.select().where(blog.c.id == 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
2
3from ariadne import QueryType, convert_kwargs_to_snake_case
4
5from crud import get_blogs, get_blog # Create a file called crud.py and add the get_blogs function
6from schemas.error import MyGraphQLError
7
8
9@convert_kwargs_to_snake_case
10async def resolve_blogs(obj, info, skip: Optional[int] = 0, limit: Optional[int] = 100):
11 blogs = await get_blogs(skip=skip, limit=limit)
12
13 return {"blogs": blogs}
14
15
16@convert_kwargs_to_snake_case
17async def resolve_blog(obj, info, blog_id):
18 blog = await get_blog(blog_id=blog_id)
19
20 if not blog:
21 raise MyGraphQLError(code=404, message=f"blog id {blog_id} not found")
22
23 return {"success": True, "blog": blog}
24
25
26query = QueryType()
27query.set_field("blogs", resolve_blogs)
28query.set_field("blog", resolve_blog)

Exactly

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, subscriptions.py and define our subscription source in it as follows:

1from ariadne import SubscriptionType, convert_kwargs_to_snake_case
2from graphql.type import GraphQLResolveInfo
3
4subscription = SubscriptionType()
5
6
7@convert_kwargs_to_snake_case
8@subscription.field("reviewblog")
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
2@convert_kwargs_to_snake_case
3@subscription.source("reviewblog")
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")
8
9 while True:
10 blog_id = await rabbit.consumeblog()
11 if blog_id:
12 yield blog_id
13 else:
14 return

Winner

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 }
13}

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 )
7}
  • 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 }
18}

Conclusion

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!

this-is-only-the-beginning-just-the-beginning

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

ABOUT US

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.