Getting started with GraphQL in Python with FastAPI and Ariadne

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:

python --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:

fastapi[uvicorn]
ariadne

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:

async def hello_world():
    return "Hello world"

• Using await, we would call our asynchronous function as follows:

obytes = 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.

type User {
    id: ID! // This is the id of the user
    email: String! // This is the email of the user
    password: String! // This is the password of the user
}

type blog {
    id: ID! // This is the id of the blog
    title: String! // This is the title of the blog
    description: String! // This is the description of the blog
    completed: Boolean! // This is the completed status of the blog
    ownerId: ID! // This is the id of the owner of the blog
}

type blogResult {
    errors: [String] // This is the list of errors
    blog: blog // This is the blog
}

type blogsResult {
    errors: [String]
    blogs: [blog] // This is the list of blogs
}

type InsertResult {
    errors: [String]
    id: ID // This is the id of the inserted blog
}

type TokenResult {
    errors: [String]
    token: String // This is the token
}

• After Define the schema, lets add the Query, Mutation, Subscription and Type definitions.

schema {
    query: Query // This is the query type
    mutation: Mutation // This is the mutation type
    subscription: Subscription // This is the subscription type
}

type Query {
    blogs: blogsResult! // This is the list of blogs
    blog(blogId: ID!): blogResult! // This is the blog
}

type Mutation {
    createblog(title: String!, description: String!): InsertResult! // This is the blog
    createUser(email: String!, password: String!): InsertResult! // This is the user
    createToken(email: String!, password: String!): TokenResult! // This is the token
}

type Subscription {
    reviewblog(token:String!): InsertResult! // This is the blog
}

Setting up the project

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

from ariadne import QueryType, make_executable_schema, load_schema_from_path
from ariadne.asgi import GraphQL

type_defs = load_schema_from_path("schema.graphql")

query = QueryType()


@query.field("hello")
def resolve_hello(*_):
    return "Hello world!"


schema = make_executable_schema(type_defs, query)
app = 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:

uvicorn app:app --reload

• Open the GraphQL PlayGround by visiting http://localhost:8000. Paste the hello query below and hit the “Play” button:

query {
  hello
}

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.

users = {}
blog = []
queues = []

Define API & GraphQL

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.

from ariadne import ObjectType, convert_kwargs_to_snake_case

from store import users, blogs

mutation = ObjectType("Mutation")


@mutation.field("createUser")
@convert_kwargs_to_snake_case
async def resolve_create_user(obj, info, email, password):
    try:
        if not users.get(username):
            user = {
                "id": len(users) + 1,
                "email": email,
                "password": password,
            }
            users[username] = user
            return {
                "success": True,
                "user": user
            }
        return {
            "success": False,
            "errors": ["Username is taken"]
        }

    except Exception as error:
        return {
            "success": False,
            "errors": [str(error)]
        }

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.

@mutation.field("createblog")
@convert_kwargs_to_snake_case
async def resolve_create_blog(obj, info, content, title, description, completed, ownerId):
    try:
        blog = {
            "ID": id,
            "title": title,
            "description": description
            "completed": completed,
            "ownerId": ownerId
        }
        blogs.append(blog)
        return {
            "success": True,
            "blog": blog
        }
    except Exception as error:
        return {
            "success": False,
            "errors": [str(error)]
        }

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:

from mutations import mutation

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

schema = make_executable_schema(type_defs, query, mutation)

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.

# as you know here i use Database[PostgreSQL] to connect to the database
# Install psycopg2 & databases[postgresql] & asyncpg

async def get_blogs(
    skip: Optional[int] = 0, limit: Optional[int] = 100
) -> Optional[Dict]:
    query = blog.select(offset=skip, limit=limit)
    result = await database.fetch_all(query=query)
    return [dict(blog) for blog in result] if result else None

async def get_blog(blog_id: int) -> Optional[Dict]:
    query = blog.select().where(blog.c.id == int(blog_id))
    result = await database.fetch_one(query=query)
    return dict(result) if result else None

• then we could use it in our schema:

from typing import Optional

from ariadne import QueryType, convert_kwargs_to_snake_case

from crud import get_blogs, get_blog # Create a file called crud.py and add the get_blogs function
from schemas.error import MyGraphQLError


@convert_kwargs_to_snake_case
async def resolve_blogs(obj, info, skip: Optional[int] = 0, limit: Optional[int] = 100):
    blogs = await get_blogs(skip=skip, limit=limit)

    return {"blogs": blogs}


@convert_kwargs_to_snake_case
async def resolve_blog(obj, info, blog_id):
    blog = await get_blog(blog_id=blog_id)

    if not blog:
        raise MyGraphQLError(code=404, message=f"blog id {blog_id} not found")

    return {"success": True, "blog": blog}


query = QueryType()
query.set_field("blogs", resolve_blogs)
query.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, subscriptions.py and define our subscription source in it as follows:

from ariadne import SubscriptionType, convert_kwargs_to_snake_case
from graphql.type import GraphQLResolveInfo

subscription = SubscriptionType()


@convert_kwargs_to_snake_case
@subscription.field("reviewblog")
async def review_blog_resolver(review_blog, info: GraphQLResolveInfo, token: str):
    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.

# This is Example from My Project FastQL
@convert_kwargs_to_snake_case
@subscription.source("reviewblog")
async def review_blog_source(obj, info: GraphQLResolveInfo, token: str):
    user = await security.get_current_user_by_auth_header(token)
    if not user:
        raise MyGraphQLError(code=401, message="User not authenticated")

    while True:
        blog_id = await rabbit.consumeblog()
        if blog_id:
            yield blog_id
        else:
            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.

mutation {
  createUser(
      email:"[e-mail]"
      password:"admin"
  ) {
    success
    user {
      userId
      email
      password
    }
  }
}

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.

query {
  userId(
      // User One
      email: "[e-mail]"
      password: "admin"
  )
}

• Publish your blog to the second user by using the createBlog mutation.

mutation {
  createBlog(
    senderId: "1",
    recipientId: "2",
    title:"Blog number1"
    description:"This is the first blog"
    ownerId: "1"
  ) {
    success
    blog {
        title
        description
        ownerId
        recipientId
        senderId
    }
  }
}

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!

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.