Asynchronous I/O in Python

asyncio is a library to write concurrent code using the async/await syntax.

Asynchronous code is non-blocking. For example, your server may need to reach out to a database or external API. This could take some time. Therefore, the code bookmarks the task and moves on to the next line of code. Or, better yet, it handles the next request.

This is a wonderful feature but it can be confusing because our code will not execute procedurally. We may expect a variable to have a value, but the variable may not be defined yet by the time the code (further down the file) runs. To handle the execution of asynchronous code, programming languages use the async/await syntax.

Here is an example from SQL Alchemy's documentation of Python's asyncio library being used with SQLAlchemy for Object Relation Models (ORM).

import asyncio

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.ext.asyncio import AsyncSession

async def async_main():
    engine = create_async_engine(
        "postgresql+asyncpg://scott:tiger@localhost/test", echo=True,
    )
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

    async with AsyncSession(engine) as session:
        async with session.begin():
            session.add_all(
                [
                    A(bs=[B(), B()], data="a1"),
                    A(bs=[B()], data="a2"),
                    A(bs=[B(), B()], data="a3"),
                ]
            )

        stmt = select(A).options(selectinload(A.bs))

        result = await session.execute(stmt)

        for a1 in result.scalars():
            print(a1)
            for b1 in a1.bs:
                print(b1)

        result = await session.execute(select(A).order_by(A.id))

        a1 = result.scalars().first()

        a1.data = "new data"

        await session.commit()

asyncio.run(async_main())

This can be really complicated if you don't understand the syntax, but it's an example of the library being used in the wild in a practical situation.

But what are all these await and asyncs doing?

Coroutines and Tasks

Using async to define a function makes the function a coroutine. There's a lower level explanation of asyncio on StackOverflow that defines coroutines as,

...functions that can be stopped and resumed while being run. In Python, they are defined using the async def keyword.

When we create a coroutine, we add functionality that controls how we can execute the function.

async def coroutine_example():
  # do stuff

To run an asynchronous program we need to use asyncio.run(coroutine).

import asyncio

async def main():
    print('Hello')
    print('world!')
    
asyncio.run(main())

If we don't use the run() method we get an error.

Await

Let's modify the program to create a coroutine inside of a coroutine. This is similar to what is happening in the SQL Alchemy example.

import asyncio

async def main():

    async def coroutine_example():
        print('Hello')
        print('world!')

    coroutine_example()
    
asyncio.run(main())

However, this will fail.

RuntimeWarning: coroutine 'main.<locals>.coroutine_example' was never awaited

To get the function to run we have to add the await keyword in front.

import asyncio

async def main():

    async def coroutine_example():
        print('Hello')
        print('world!')

    await coroutine_example()
    
asyncio.run(main())

Coroutines and Tasks are both awaitable objects. That means that we can call them using the await keyword.

Tasks

Tasks wrap coroutines and expose new methods. We create Tasks by passing in a coroutine to asyncio.create_task(coroutine).

import asyncio

async def main():

    async def coroutine_example():
        print('Hello')
        print('world!')

    task1 = asyncio.create_task(coroutine_example())

    await task1
    
asyncio.run(main())

We can now call methods on task1 like cancel(), done(), result(), etc.

So far the changes in this file have been trivial. We have just added wrappers around a very simple program. In the final section we'll look at running tasks concurrently.

`asyncio.gather()`

Next, let's break the program up into two coroutines.

import asyncio

async def main():

    async def hello():
        print('Hello')

    async def world():
        print('world!')

    await hello()
    await world()
    
    
asyncio.run(main())

Still, nothing special. To run these functions at the same time we can pass them to the gather() method.

import asyncio

async def main():

    async def hello():
        print('Hello')

    async def world():
        print('world!')

    await asyncio.gather(
        hello(),
        world()
        )
    
    
asyncio.run(main())

Peeking at the Python documentation for gather we read:

Run awaitable objects in the aws sequence concurrently.

If any awaitable in aws is a coroutine, it is automatically scheduled as a Task.

If all awaitables are completed successfully, the result is an aggregate list of returned values. The order of result values corresponds to the order of awaitables in aws.

aws is short for awaitables.

In the following code, we use gather() to call the hello() function twice, while only calling world() once.

import asyncio

async def main():

    async def hello(order):
        print(f'{(order)} Hello')

    async def world():
        await asyncio.sleep(3)
        print('world!')

    await asyncio.gather(
        hello(1),
        world(),
        hello(3)
        )
    
    
asyncio.run(main())

Output:

1 Hello
3 Hello
world!

If these functions were dependent on eachother, we could go back to using the await keyword and make the calls individually.

import asyncio

async def main():

    async def hello(order):
        print(f'{(order)} Hello')

    async def world():
        await asyncio.sleep(3)
        print('world!')

    await hello(1)
    await world()
    await hello(3)
    
    
asyncio.run(main())