Python Async IO

AsyncIO is a concurrent programming design that has received dedicated support in Python. AsyncIO is a built-in Python library used to write concurrent, asynchronous, and cooperative code in a sequential style. A program is called concurrent when multiple tasks are running and overlapping each other’s lifecycle. In other words, we can switch from one task to another task for execution and then switch back to it concurrently. Whereas in parallelism the tasks run simultaneously.

Introduction to Async IO
Async Recap
Convert a sync program to async program
Async to the Next Level
References

Introduction to Async IO

Parallelism consists of performing multiple operations at the same time. Multiprocessing is a means to effect parallelism, and it entails spreading tasks over a computer’s central processing units (CPUs, or cores). Multiprocessing is well-suited for CPU-bound tasks: tightly bound for loops and mathematical computations usually fall into this category.
Concurrency is a slightly broader term than parallelism. It suggests that multiple tasks have the ability to run in an overlapping manner. (There’s a saying that concurrency does not imply parallelism.)
Threading is a concurrent execution model whereby multiple threads take turns executing tasks. One process can contain multiple threads. Python has a complicated relationship with threading thanks to its GIL.

Async IO is a style of concurrent programming, but it is not parallelism. It’s more closely aligned with threading than with multiprocessing but is very much distinct from both of these and is a standalone member in concurrency’s bag of tricks.

Asynchronous routines are able to pause while waiting on their ultimate result and let other routines run in the meantime.

Async Recap

async and await are the core keywords that make up the async IO framework in Python.

async on the function and await on the function or method call that the next statement will wait for the result of the call. asyncio.run() to start the event loop

import asyncio

async def count():
    print("One")
    await asyncio.sleep(1)
    print("Two")

async def main():
    await asyncio.gather(count(), count(), count())

if __name__ == "__main__":
    import time
    s = time.perf_counter()
    asyncio.run(main())
    elapsed = time.perf_counter() - s
    print(f"{__file__} executed in {elapsed:0.2f} seconds.")

The result of the above code is:

$ python3 countasync.py
One
One
One
Two
Two
Two
countasync.py executed in 1.01 seconds.

The order of this output is the heart of async IO. Talking to each of the calls to count() is a single event loop, or coordinator. When each task reaches await asyncio.sleep(1), the function yells up to the event loop and gives control back to it, saying, I’m going to be sleeping for 1 second. Go ahead and let something else meaningful be done in the meantime.”

await asyncio.gather(count(), count(), count()) is a function that takes in a variable number of awaitables and schedules them to run concurrently. It returns a single awaitable that waits for all of them to complete.

Convert a sync program to async program

Helper function to call a url in sync mode

# req_http.py
import asyncio
import requests

# A few handy JSON types
JSON = int | str | float | bool | None | dict[str, "JSON"] | list["JSON"]
JSONObject = dict[str, JSON]
JSONList = list[JSON]

async def http_get(url: str) -> JSONObject:
    '''function to call a url in async mode, get the response and returns a JSON object'''
    return await asyncio.to_thread(http_get_sync, url)

# main.py
import asyncio
from random import randint
from req_http import JSONObject, http_get

MAX_POKEMON = 898

async def get_pokemon(pokemon_id: int) -> JSONObject:
    pokemon_url = f"https://pokeapi.co/api/v2/pokemon/{pokemon_id}"
    return await http_get(pokemon_url)

async def main() -> None:
    pokemon_id = randint(1, MAX_POKEMON)
    pokemon = await get_pokemon(pokemon_id + 1)
    print(pokemon["name"])

if __name__ == "__main__":
    asyncio.run(main())

Async to the Next Level

import asyncio
import time
import requests


async def counter(until: int = 10) -> None:
    now = time.perf_counter()
    print("Started counter")
    for i in range(0, until):
        last = now
        await asyncio.sleep(0.01)
        now = time.perf_counter()
        print(f"{i}: Was asleep for {now - last}s")


def send_request(url: str) -> int:
    print("Sending HTTP request")
    response = requests.get(url)
    return response.status_code


async def send_async_request(url: str) -> int:
    return await asyncio.to_thread(send_request, url)


async def main() -> None:
    status_code, _ = await asyncio.gather(
        send_async_request("https://www.arjancodes.com"), counter()
    )
    print(f"Got HTTP response with status {status_code}.")


if __name__ == "__main__":
    asyncio.run(main())