AsyncIO in Python

First of all, it's worth getting familiar with the architecture of the event loop. Its general scheme looks as follows:

Elements of the event loop

The event loop includes the following elements:

1. Selector — listens for events from the operating system and hands work off to coroutines that are waiting for IO messages to be processed.

2. IO tasks — a special task is added to the scheduler to handle IO events from the operating system.

3. Scheduler — processes tasks in the task queue and makes sure they switch between one another correctly. The key element of the whole program.

4. System Call — blocks of code that extend the scheduler's functionality.

5. Task Queue — new tasks to be executed are collected in this queue.

6. Task — the main unit of work in the program's loop. A task stores information about the coroutine being run. It can handle a chain of nested coroutines.

7. Coroutine — executable code that the task scheduler operates on.

A simple program

Let's look at the simplest Python program that uses asyncio.

PythonCopy

import asyncio


async def calc(n: int):
    result = n ** 2
    await asyncio.sleep(2)
    return result


async def run(n: int):
    result = await calc(n)
    print(result)


if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    asyncio.run(run(2))
    loop.close()

Changes in Python 3.11

It's worth noting right away that in earlier versions of Python you could simply get the current event loop via the asyncio.get_event_loop() function and then pass a coroutine to it through loop.run_until_complete(func_coro()). But starting with Python 3.11 such code raises an error.