Python’s asyncio library is the built-in Python library for running code concurrently with the async/await keywords. The asyncio library is ideal for IO bound and structured network code. The difference between when to use the run command and the run_until_complete command with a loop is subtle but could have real implications for your code. In this post we’ll go over:

What is AsyncIO Run?
- asyncio.run example
What is AsyncIO’s Run Until Complete?
- asyncio.loop.run_until_complete example
A Commentary on AsyncIO loops
When to Use AsyncIO Run vs AsyncIO Loop’s Run Until Complete

AsyncIO Run – asyncio.run()

asyncio.run() is a high-level API in the asyncio library. It takes one required parameter, a coroutine, and one default named parameter, debug, which is set to False. The coroutine passed into asyncio.run() must be an async/await function. As I said earlier, this is a high-level API in the asyncio library, and under the hood it actually calls loop.run_until_complete().

This is the recommended entry point for asynchronous programs and should only be called once. The run function creates, executes, and closes a loop to run the run_until_complete function.

asyncio.run() example

We’ll use a simple example of waiting for 1 second using the asyncio.sleep() function. We’ll create an async test function that takes no parameters. All it does is print out the current time, await a sleep of 1, and then print out the current time again. Then we’ll use the async.run() function to test it.

import asyncio
import time
 
async def test():
    print(time.time())
    await asyncio.sleep(1)
    print(time.time())
 
asyncio.run(test())

It should print out something like this:

AsyncIO Run Until Complete – asyncio.loop.run_until_complete()

asyncio.loop.run_until_complete() is a low-level API in the asyncio native Python library. The function takes one parameter, a future object. A coroutine or task defined by the async/await keywords is a future object. That’s why we can call asyncio.loop.run_until_complete() with asyncio.run().

In order to use the asyncio.loop object’s run_until_complete() function, we need to actually have an asynio.loop object already instantiated or instantiate one on the call.

asyncio.loop.run_until_complete() example

We will use the exact same function we used above in our example of asyncio.run(). We’ll create a function that prints the time, sleeps for a second, and then prints the time again. The only difference is how we’re going to run it.

In this example, we’re going to use asyncio.new_event_loop() to create an async.loop object and then call run_until_complete() on the test() function. We should get the exact same result.

import asyncio
import time
 
async def test():
    print(time.time())
    await asyncio.sleep(1)
    print(time.time())
 
asyncio.new_event_loop().run_until_complete(test())

It should print out something like this:

asyncio run until complete sleep one second test

Comment on AsyncIO Loops

Event loops are the core of asynchronous functions in Python. They run tasks, callbacks, network I/O, and subprocesses. Event loops are not meant to be used directly. They can be, but they shouldn’t be. They interact with asyncio’s low-level APIs.

When to use asyncio.run vs asyncio.loop.run_until_complete

You should almost always use asyncio.run(). The only time you should use the run_until_complete() function over run() is if you need access to other low-level APIs. Modern applications are suggested to use asyncio.run().

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.