Send API Requests Asynchronously in Python

Sending HTTP API requests in Python is simple. We can just use the requests library like we do in so many examples such as when we built our AI Content Moderation System, explored NLP libraries in Python, and Summarized November 2021 in Headlines. What if we could do it faster though? Using asyncio and aiohttp, we can use Python to send asynchronous requests.

An asynchronous request is one that we send asynchronously instead of synchronously. This means we can do non I/O blocking operations separately. This allows us to speed up our Python program.

In this post we’re going to go over:

• When and why you should use asynchronous requests in Python
• The Python libraries needed for asynchronous requests
• Creating an asynchronous function to run requests concurrently
  • Creating a Semaphore task
  • Returning gathered tasks
• Creating asynchronous API calls
• Configuring example API calls 
• Calling multiple APIs asynchronously example
• Testing multiple API calls asynchronously example

To follow this tutorial you’ll need to install the aiohttp library. The asyncio and json libraries are native to Python. You can install aiohttp with the command below.

pip install aiohttp

When and Why Use Async Requests in Python?

You don’t always need to use asynchronous requests in Python. Asynchronous requests are only useful in two general cases. One, is if you have multiple requests to run. Two is if you have non I/O blocking operations to do and you don’t need the result of your request until much later. In this post, we will be showcasing example one by running three API requests concurrently.

Import Required Python Libraries for Asynchronous Requests

The aiohttp library is the main driver of sending concurrent requests in Python. The asyncio library is a native Python library that allows us to use async and await in Python. These are the basics of asynchronous requests. The other library we’ll use is the `json` library to parse our responses from the API. To do this specific tutorial, you’ll need to sign up for a free API key at The Text API. Lastly, we’ll import our API key. You can save it anywhere, but I’ve saved it in a text_api_config module.

import asyncio
import aiohttp
import json
 
from text_api_config import apikey

Create Async Function to Run Requests Concurrently

The first function we’re going to create is the function we’re going to use to run multiple asynchronous requests concurrently. For this example, we’re going to do this with the Semaphore object from asyncio. A semaphore is essentially a way to hold an object and share it between threads. This is NOT a threadsafe object.

Create Semaphore Task with asyncio

We’ll create an asynchronous function that takes two parameters, n, the number of concurrent “threads”, and *tasks, a variable list of tasks. The function will start by creating a semaphore object with n “threads” in it. Then we will create an async function inside of our already asynchronous function that uses the semaphore object that will await a task.

async def gather_with_concurrency(n, *tasks):
    semaphore = asyncio.Semaphore(n)
    async def sem_task(task):
        async with semaphore:
            return await task

Once we’ve created our semaphore object, the only thing left to do is use it on all the tasks. We’ll use asyncio.gather to run the async semaphore object on each task in our task list. Notice the star outside of the parenthesis defining the tuple object of sem_task for each task? That’s for “unpacking” the tuple so that our gather function is able to execute correctly.

    return await asyncio.gather(*(sem_task(task) for task in tasks))

Full Code for Gathering Asynchronous Python Requests to Run Concurrently

Here’s the full code for creating a function that runs multiple asynchronous tasks concurrently.

async def gather_with_concurrency(n, *tasks):
    semaphore = asyncio.Semaphore(n)
    async def sem_task(task):
        async with semaphore:
            return await task
   
    return await asyncio.gather(*(sem_task(task) for task in tasks))

Create Python Asynchronous API Call Function

Earlier we created our function to run multiple asynchronous calls at once, now let’s create a function to make asynchronous calls. We’ll create another async function that will take four parameters, a url API endpoint, a session object, which will help speed up our function, headers, and a body.

All we need to do is use the session to send a POST request. If you’ve seen the other posts including requests like building your own AI Text Summarizer, you’ll notice that we get the text from the response differently than in this case. In this case we’re calling text() as a function instead of text as a parameter. That’s because of the way that async is naturally built. We delay loading the text immediately, we do it asynchronously. That’s what we have to await it. At the end of our function we’ll return the JSON of our request text.

async def post_async(url, session, headers, body):
    async with session.post(url, headers=headers, json=body) as response:
        text = await response.text()
        return json.loads(text)

Configure Asynchronous Python API Calls

Now that we’ve built the functions to run asynchronous requests concurrently and send asynchronous requests, let’s configure the API calls. We’re using three example API endpoints from The Text API, for more information check out the documentation.

First, we’ll set up headers that will tell the server that we’re sending a JSON object and pass in the API key. Then we’ll send it a body. I just wrote a random body that gives commentary on myself and PythonAlgos. Finally, we’ll set up the three URLs that are our API endpoints.

headers = {
    "Content-Type": "application/json",
    "apikey": apikey
}
body = {
    "text": "Yujian Tang is the best software content creator. PythonAlgos is the best and fastest way to learn Python and software skills. Tell your friends!"
}
url = "https://app.thetextapi.com/text/"
summarize_url = url+"summarize"
ner_url = url+"ner"
mcp_url = url+"most_common_phrases"

Call Multiple APIs Asynchronously in Python with AIOHTTP

Now let’s get to the main function of our program. We’re going to use the TCPConnector and ClientSession objects from aiohttp to do the heavy lifting. In this function we’re going to create a session, and use it with the two functions we created above to call the three API endpoints we defined concurrently.

Setup HTTP Session with aiohttp

The first thing we’re going to do in our main function is set up the HTTP session. First, we’ll create a TCPConnector object. Then, we’ll use that connector object to create a ClientSession object. We will also make our list of URLs and define the number of concurrent requests here.

async def main():
    conn = aiohttp.TCPConnector(limit=None, ttl_dns_cache=300)
    session = aiohttp.ClientSession(connector=conn)
    urls = [summarize_url, ner_url, mcp_url]
    conc_req = 3

Execute Gathered Requests with aiohttp

After setting up our HTTP session, let’s use the gather_with_concurrency and post_async functions to send our three API requests concurrently. After we send the request, we’ll simply close the session and print out the results.

    summary, ner, mcp = await gather_with_concurrency(conc_req, *[post_async(url, session, headers, body) for url in urls])
    await session.close()
    print(summary["summary"])
    print(ner["ner"])
    print(mcp["most common phrases"])

Full Code Example Using aiohttp for Async Python Requests

Here’s the full code for the main function to create a session and concurrently call three API requests with the functions created above.

async def main():
    conn = aiohttp.TCPConnector(limit=None, ttl_dns_cache=300)
    session = aiohttp.ClientSession(connector=conn)
    urls = [summarize_url, ner_url, mcp_url]
    conc_req = 3
    summary, ner, mcp = await gather_with_concurrency(conc_req, *[post_async(url, session, headers, body) for url in urls])
    await session.close()
    print(summary["summary"])
    print(ner["ner"])
    print(mcp["most common phrases"])

Testing Python Asynchronous Requests with asyncio and aiohttp

To test our function, we use asyncio and call get_event_loop() and run_until_comiplete() on our function. 

asyncio.get_event_loop().run_until_complete(main())

We should see an output like the one below.

Further Reading

• Nested Lists in Python
• Using Funcanimation in Python
• Build Your Own AI Text Summarizer
• Create a Low Level Design Document
• How to do Math on Python

I run this site to help you and others like you find cool projects and practice software skills. If this is helpful for you and you enjoy your ad free site, please help fund this site by donating below! If you can’t donate right now, please think of us next time.