In a previous article, we covered how to create a basic API using GraphQL and Python. In this mini-project below, we’ll use a GraphQL feature called subscriptions to create a functional chat application.

What is GraphQL Subscription?

Subscriptions is a GraphQL functionality that allows the GraphQL server to send data to the client whenever a specific event occurs. Here, the subscriptions remain in the connected state with the client, removing the request-response cycle of the API.

GraphQL subscriptions let you create real-time applications using GraphQL APIs that return only data that the client wants. It’s kinda similar to websockets. Subscriptions are event-driven, meaning if there are any updates on the server-side, the client is automatically notified about the update.

Now that we have a fundamental knowledge of Subscriptions, let’s focus on our mini-project:

Prerequisites:

• Python
• Basic knowledge of GraphQL

Step –1: Environment Setup

Open your terminal and create a new folder using the below command:

mkdir chatql

Then open the folder:

cd chatql

Now, once you’re in the folder, let’s initialize a virtual environment.

virtualenv .

Now activate it:

source bin/activate

Now let’s install a few dependencies. So, run the below command:

pip3 install ariadne "uvicorn[standard]"

Now we’re done with the environment setup. Let’s proceed with the next steps.

Step –2: Define Mutations

Create a file name mutations.py and paste the below code:

from ariadne import ObjectType, convert_kwargs_to_snake_case

from store import users, messages, queues

mutation = ObjectType("Mutation")


@mutation.field("createUser")
@convert_kwargs_to_snake_case
async def resolve_create_user(obj, info, username):
    try:
        if not users.get(username):
            user = {
                "user_id": len(users) + 1,
                "username": username
            }
            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)]
        }

@mutation.field("createMessage")
@convert_kwargs_to_snake_case
async def resolve_create_message(obj, info, content, sender_id, recipient_id):
    try:
        message = {
            "content": content,
            "sender_id": sender_id,
            "recipient_id": recipient_id
        }
        messages.append(message)
        for queue in queues:
            await queue.put(message)
        return {
            "success": True,
            "message": message
        }
    except Exception as error:
        return {
            "success": False,
            "errors": [str(error)]
        }

Code Explanation:

We have defined two mutations: createUser and createMessage. As the names suggest, createUser creates users while createMessage creates messages. The createUser takes one parameter, the username, and returns the id of the user. If the user already exists, it returns an error. The createMessage mutation takes the message, sender_id, and recipient_id.

Step –3: Define Schema

Create a file named schema.graphql and paste the below code:

type User {
    username: String
    userId: String
}

type Message {
    content: String
    senderId: String
    recipientId: String
}

type createUserResult {
    user: User
    success: Boolean!
    errors: [String]
}

type createMessageResult {
    message: Message
    success: Boolean!
    errors: [String]
}

type messagesResult {
    messages: [Message]
    success: Boolean!
    errors: [String]
}

type Query {
    hello: String!
    messages(userId: String!): messagesResult
    userId(username: String!): String
}

type Mutation {
    createUser(username: String!): createUserResult
    createMessage(senderId: String, recipientId: String, content: String): createMessageResult
}


type Subscription {
    messages(userId: String): Message
}

schema {
  query: Query
  mutation: Mutation
  subscription: Subscription
}

Code Explanation:

We have defined type User, Message, createUserResult, createMessageResult, and messagesResult. The User has data type defined username and userId as string. The Message has data types defined as content, senderId and recipientId as string.

We also have a query, mutation, and subscription, which will be used to resolve the query and enable subscription.

Step –4: Defining Subscriptions

Create a new file subscriptions.py and paste the below code:

import asyncio
from ariadne import convert_kwargs_to_snake_case, SubscriptionType

from store import queues

subscription = SubscriptionType()


@subscription.source("messages")
@convert_kwargs_to_snake_case
async def messages_source(obj, info, user_id):
    queue = asyncio.Queue()
    queues.append(queue)
    try:
        while True:
            print('listen')
            message = await queue.get()
            queue.task_done()
            if message["recipient_id"] == user_id:
                yield message
    except asyncio.CancelledError:
        queues.remove(queue)
        raise

@subscription.field("messages")
@convert_kwargs_to_snake_case
async def messages_resolver(message, info, user_id):
    return message

Code Explanation:

The messaging system here will follow First In First Out (FIFO). This means that the first message that comes in the queue will be sent to the client. There’s an infinite loop running, which will check for new messages infinitely using await queue.get().

Step –5: Defining Queries

Create a new file queries.py and paste the below code:

from ariadne import ObjectType, convert_kwargs_to_snake_case

from store import messages, users

query = ObjectType("Query")


@query.field("messages")
@convert_kwargs_to_snake_case
async def resolve_messages(obj, info, user_id):
    def filter_by_userid(message):
        return message["sender_id"] == user_id or \
               message["recipient_id"] == user_id

    user_messages = filter(filter_by_userid, messages)
    return {
        "success": True,
        "messages": user_messages
    }

@query.field("userId")
@convert_kwargs_to_snake_case
async def resolve_user_id(obj, info, username):
    user = users.get(username)
    if user:
        return user["user_id"]

Code Explanation:

We’re importing ariadne here, which we’ll use to make GraphQL implementation easier. Then we have defined two query fields, messages and userId, which will fetch the messages for a user id and user details like username, respectively.

We also imported a store module, which is used to store messages. We’ll discuss this module in the next steps.

Step –6: Defining the store.py and app.py Module:

This is the last step of the setup. So create a file store.py and paste the below code:

users = dict()
messages = []
queues = []

Code Explanation:

We have created a dictionary of users, a list of messages, and queues.

Now create another file, app.py, and paste the below code:

from ariadne import make_executable_schema, load_schema_from_path, \
    snake_case_fallback_resolvers
from ariadne.asgi import GraphQL
from mutations import mutation
from queries import query
from subscriptions import subscription


type_defs = load_schema_from_path("schema.graphql")

schema = make_executable_schema(type_defs, query, mutation, subscription,
                                snake_case_fallback_resolvers)
app = GraphQL(schema, debug=True)

Code Explanation:

We’re importing all the modules that we created, i.e., mutations, queries, subscriptions, and also ariadne. The code is loading schema.graphql and storing in type_defs. Then we’re also calling make_executable_schema() and passing mutations, query, subscriptions, resolvers, and type_defs. This function is used to create an executable schema.

Now it’s time to test the API.

Step –7: Testing the API

To run the server, use the below command:

uvicorn app:app --reload

Now open your web browser and head over to URL: https://127.0.0.1:8000/ and paste the below code on the left pane of the page to create a new user:

mutation {
  createUser(username:"YOUR_NAME_1_HERE") {
    success
    user {
      userId
      username
    }
  }
}

We’re creating a user here. Just replace the placeholder YOUR_NAME_1_HERE with a unique username. Make sure you create two users because you have to send a message to someone, right? Once you execute the above query, it’ll return something like this:

{
  "data": {
    "createUser": {
      "success": true,
      "user": {
        "userId": "1",
        "username": "YOUR_NAME_1_HERE"
      }
    }
  }
}

Make sure to note down the userId of both users; we’ll need these IDs to send messages. Now open one more tab on your browser and go to URL: https://localhost:8000/. You’ll see the same page, now paste the below code to subscribe to messages:

subscription {
  messages(userId: "YOUR SECOND USER’s ID HERE") {
    content
    senderId
    recipientId
  }
}

If everything is fine, you should see something like this:

Can you see “Listening” text at the bottom of the screen? It means that it has subscribed to the messages.

Now go back to the previous tab where you opened the URL: https://127.0.0.1:8000/ and on the left pane replace the existing code with the below code:

mutation {
  createMessage(
    senderId: "YOUR FIRST USER’s ID HERE",
    recipientId: "YOUR SECOND USER’s ID HERE",
    content:"YOUR MESSAGE HERE"
  ) {
    success
    message {
      content
      recipientId
      senderId
    }
  }
}

Just replace the placeholders with the correct IDs and the message you’d like to send, and hit the play button. On the second tab, you now should be able to receive the messages. That’s it! Your chat API is ready now.

Final Words:

We have covered a basic chat application API in this article using Python and GraphQL. You can create a UI using a JS framework and make it more interactive, while you can also use JWT to make it more secure.