What’s New in This Chapter

When I wrote the first edition of Fluent Python, the asyncio library was provisional and the async/await keywords did not exist. Therefore, I had to update all examples in this chapter. I also created new examples: domain probing scripts, a FastAPI web service, and experiments with Python’s new asynchronous console mode.

The ideas in “How Async Works and How It Doesn’t” reflect hard-earned lessons that I consider essential reading for anyone using asynchronous programming. They may save you a lot of trouble—whether you’re using Python or Node.js.

Finally, I removed several paragraphs about asyncio.Futures, which is now considered part of the low-level asyncio APIs.

An asyncio Example: Probing Domains

Imagine you are about to start a new blog on Python, and you plan to register a domain using a Python keyword and the .DEV suffix—for example: AWAIT.DEV. Example 21-1 is a script using asyncio to check several domains concurrently. This is the output it produces:

$ python3 blogdom.py
  with.dev
+ elif.dev
+ def.dev
  from.dev
  else.dev
  or.dev
  if.dev
  del.dev
+ as.dev
  none.dev
  pass.dev
  true.dev
+ in.dev
+ for.dev
+ is.dev
+ and.dev
+ try.dev
+ not.dev

Note that the domains appear unordered. If you run the script, you’ll see them displayed one after the other, with varying delays. The + sign indicates your machine was able to resolve the domain via DNS. Otherwise, the domain did not resolve and may be available.⁵

In blogdom.py, the DNS probing is done via native coroutine objects. Because the asynchronous operations are interleaved, the time needed to check the 18 domains is much less than checking them sequentially. In fact, the total time is practically the same as the time for the single slowest DNS response, instead of the sum of the times of all responses.

Example 21-1 shows the code for blogdom.py.

#!/usr/bin/env python3
import asyncio
import socket
from keyword import kwlist

MAX_KEYWORD_LEN = 4  


async def probe(domain: str) -> tuple[str, bool]:  
    loop = asyncio.get_running_loop()  
    try:
        await loop.getaddrinfo(domain, None)  
    except socket.gaierror:
        return (domain, False)
    return (domain, True)


async def main() -> None:  
    names = (kw for kw in kwlist if len(kw) <= MAX_KEYWORD_LEN)  
    domains = (f'{name}.dev'.lower() for name in names)  
    coros = [probe(domain) for domain in domains]  
    for coro in asyncio.as_completed(coros):  
        domain, found = await coro  
        mark = '+' if found else ' '
        print(f'{mark} {domain}')


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

Set maximum length of keyword for domains, because shorter is better.

probe returns a tuple with the domain name and a boolean; True means the domain resolved. Returning the domain name will make it easier to display the results.

Get a reference to the asyncio event loop, so we can use it next.

The loop.getaddrinfo(…) coroutine-method returns a five-part tuple of parameters to connect to the given address using a socket. In this example, we don’t need the result. If we got it, the domain resolves; otherwise, it doesn’t.

main must be a coroutine, so that we can use await in it.

Generator to yield Python keywords with length up to MAX_KEYWORD_LEN.

Generator to yield domain names with the .dev suffix.

Build a list of coroutine objects by invoking the probe coroutine with each domain argument.

asyncio.as_completed is a generator that yields coroutines that return the results of the coroutines passed to it in the order they are completed—not the order they were submitted. It’s similar to futures.as_completed, which we saw in Chapter 20, Example 20-4.

At this point, we know the coroutine is done because that’s how as_completed works. Therefore, the await expression will not block but we need it to get the result from coro. If coro raised an unhandled exception, it would be re-raised here.

asyncio.run starts the event loop and returns only when the event loop exits. This is a common pattern for scripts that use asyncio: implement main as a coroutine, and drive it with asyncio.run inside the if __name__ == '__main__': block.

async def probe(domain: str) -> tuple[str, bool]:
    loop = asyncio.get_running_loop()
    try:
        await loop.getaddrinfo(domain, None)
    except socket.gaierror:
        return (domain, False)
    return (domain, True)

def probe(domain: str) -> tuple[str, bool]:  # no async
    loop = asyncio.get_running_loop()
    try:
        loop.getaddrinfo(domain, None)  # no await
    except socket.gaierror:
        return (domain, False)
    return (domain, True)

The for keyword works with iterables. The await keyword works with awaitables.

However, end-user code does not always need to await on a Task. We use asyncio.create_task(one_coro()) to schedule one_coro for concurrent execution, without waiting for its return. That’s what we did with the spinner coroutine in spinner_async.py (Example 19-4). If you don’t expect to cancel the task or wait for it, there is no need to keep the Task object returned from create_task. Creating the task is enough to schedule the coroutine to run.

In contrast, we use await other_coro() to run other_coro right now and wait for its completion because we need its result before we can proceed. In spinner_async.py, the supervisor coroutine did res = await slow() to execute slow and get its result.

When implementing asynchronous libraries or contributing to asyncio itself, you may also deal with these lower-level awaitables:

• An object with an __await__ method that returns an iterator; for example, an asyncio.Future instance (asyncio.Task is a subclass of asyncio.Future)

• Objects written in other languages using the Python/C API with a tp_as_async.am_await function, returning an iterator (similar to __await__ method)

Existing codebases may also have one additional kind of awaitable: generator-based coroutine objects—which are in the process of being deprecated.

Now let’s study the asyncio version of a script that downloads a fixed set of flag images.

Downloading with asyncio and HTTPX

The flags_asyncio.py script downloads a fixed set of 20 flags from fluentpython.com. We first mentioned it in “Concurrent Web Downloads”, but now we’ll study it in detail, applying the concepts we just saw.

As of Python 3.10, asyncio only supports TCP and UDP directly, and there are no asynchronous HTTP client or server packages in the standard library. I am using HTTPX in all the HTTP client examples.

We’ll explore flags_asyncio.py from the bottom up—that is, looking first at the functions that set up the action in Example 21-2.

def download_many(cc_list: list[str]) -> int:    
    return asyncio.run(supervisor(cc_list))      

async def supervisor(cc_list: list[str]) -> int:
    async with AsyncClient() as client:          
        to_do = [download_one(client, cc)
                 for cc in sorted(cc_list)]      
        res = await asyncio.gather(*to_do)       

    return len(res)                              

if __name__ == '__main__':
    main(download_many)

This needs to be a plain function—not a coroutine—so it can be passed to and called by the main function from the flags.py module (Example 20-2).

Execute the event loop driving the supervisor(cc_list) coroutine object until it returns. This will block while the event loop runs. The result of this line is whatever supervisor returns.

Asynchronous HTTP client operations in httpx are methods of AsyncClient, which is also an asynchronous context manager: a context manager with asynchronous setup and teardown methods (more about this in “Asynchronous Context Managers”).

Build a list of coroutine objects by calling the download_one coroutine once for each flag to be retrieved.

Wait for the asyncio.gather coroutine, which accepts one or more awaitable arguments and waits for all of them to complete, returning a list of results for the given awaitables in the order they were submitted.

supervisor returns the length of the list returned by asyncio.gather.

Now let’s review the top of flags_asyncio.py (Example 21-3). I reorganized the coroutines so we can read them in the order they are started by the event loop.

import asyncio

from httpx import AsyncClient  

from flags import BASE_URL, save_flag, main  

async def download_one(client: AsyncClient, cc: str):  
    image = await get_flag(client, cc)
    save_flag(image, f'{cc}.gif')
    print(cc, end=' ', flush=True)
    return cc

async def get_flag(client: AsyncClient, cc: str) -> bytes:  
    url = f'{BASE_URL}/{cc}/{cc}.gif'.lower()
    resp = await client.get(url, timeout=6.1,
                                  follow_redirects=True)  
    return resp.read()  

httpx must be installed—it’s not in the standard library.

Reuse code from flags.py (Example 20-2).

download_one must be a native coroutine, so it can await on get_flag—which does the HTTP request. Then it displays the code of the downloaded flag, and saves the image.

get_flag needs to receive the AsyncClient to make the request.

The get method of an httpx.AsyncClient instance returns a ClientResponse object that is also an asynchronous context manager.

Network I/O operations are implemented as coroutine methods, so they are driven asynchronously by the asyncio event loop.

Your code delegates to the httpx coroutines explicitly through await or implicitly through the special methods of the asynchronous context managers, such as Async​Client and ClientResponse—as we’ll see in “Asynchronous Context Managers”.

The Secret of Native Coroutines: Humble Generators

A key difference between the classic coroutine examples we saw in “Classic Coroutines” and flags_asyncio.py is that there are no visible .send() calls or yield expressions in the latter. Your code sits between the asyncio library and the asynchronous libraries you are using, such as HTTPX. This is illustrated in Figure 21-1.

Figure 21-1. In an asynchronous program, a user’s function starts the event loop, scheduling an initial coroutine with asyncio.run. Each user’s coroutine drives the next with an await expression, forming a channel that enables communication between a library like HTTPX and the event loop.

Under the hood, the asyncio event loop makes the .send calls that drive your coroutines, and your coroutines await on other coroutines, including library coroutines. As mentioned, await borrows most of its implementation from yield from, which also makes .send calls to drive coroutines.

The await chain eventually reaches a low-level awaitable, which returns a generator that the event loop can drive in response to events such as timers or network I/O. The low-level awaitables and generators at the end of these await chains are implemented deep into the libraries, are not part of their APIs, and may be Python/C extensions.

Using functions like asyncio.gather and asyncio.create_task, you can start multiple concurrent await channels, enabling concurrent execution of multiple I/O operations driven by a single event loop, in a single thread.

In “Context Managers and with Blocks”, we saw how an object can be used to run code before and after the body of a with block, if its class provides the __enter__ and __exit__ methods.

Now, consider Example 21-4, from the asyncpg asyncio-compatible PostgreSQL driver documentation on transactions.

tr = connection.transaction()
await tr.start()
try:
    await connection.execute("INSERT INTO mytable VALUES (1, 2, 3)")
except:
    await tr.rollback()
    raise
else:
    await tr.commit()

A database transaction is a natural fit for the context manager protocol: the transaction has to be started, data is changed with connection.execute, and then a rollback or commit must happen, depending on the outcome of the changes.

In an asynchronous driver like asyncpg, the setup and wrap-up need to be coroutines so that other operations can happen concurrently. However, the implementation of the classic with statement doesn’t support coroutines doing the work of __enter__ or __exit__.

That’s why PEP 492—Coroutines with async and await syntax introduced the async with statement, which works with asynchronous context managers: objects implementing the __aenter__ and __aexit__ methods as coroutines.

With async with, Example 21-4 can be written like this other snippet from the asyncpg documentation:

async with connection.transaction():
    await connection.execute("INSERT INTO mytable VALUES (1, 2, 3)")

In the asyncpg.Transaction class, the __aenter__ coroutine method does await self.start(), and the __aexit__ coroutine awaits on private __rollback or __commit coroutine methods, depending on whether an exception occurred or not. Using coroutines to implement Transaction as an asynchronous context manager allows asyncpg to handle many transactions concurrently.

Back to flags_asyncio.py, the AsyncClient class of httpx is an asynchronous context manager, so it can use awaitables in its __aenter__ and __aexit__ special coroutine methods.

We’ll now enhance the asyncio flag download example with a progress bar, which will lead us to explore a bit more of the asyncio API.

Recall from “Downloads with Progress Display and Error Handling” that the flags2 set of examples share the same command-line interface, and they display a progress bar while the downloads are happening. They also include error handling.

For instance, Example 21-5 shows an attempt to get 100 flags (-al 100) from the ERROR server, using 100 concurrent requests (-m 100). The 48 errors in the result are either HTTP 418 or time-out errors—the expected (mis)behavior of the slow_server.py.

$ python3 flags2_asyncio.py -s ERROR -al 100 -m 100
ERROR site: http://localhost:8002/flags
Searching for 100 flags: from AD to LK
100 concurrent connections will be used.
100%|█████████████████████████████████████████| 100/100 [00:03<00:00, 30.48it/s]
--------------------
 52 flags downloaded.
 48 errors.
Elapsed time: 3.31s

Now let’s see how flags2_asyncio.py is implemented.

import asyncio
from collections import Counter
from http import HTTPStatus
from pathlib import Path

import httpx
import tqdm  # type: ignore

from flags2_common import main, DownloadStatus, save_flag

# low concurrency default to avoid errors from remote site,
# such as 503 - Service Temporarily Unavailable
DEFAULT_CONCUR_REQ = 5
MAX_CONCUR_REQ = 1000

async def get_flag(client: httpx.AsyncClient,  
                   base_url: str,
                   cc: str) -> bytes:
    url = f'{base_url}/{cc}/{cc}.gif'.lower()
    resp = await client.get(url, timeout=3.1, follow_redirects=True)   
    resp.raise_for_status()
    return resp.content

async def download_one(client: httpx.AsyncClient,
                       cc: str,
                       base_url: str,
                       semaphore: asyncio.Semaphore,
                       verbose: bool) -> DownloadStatus:
    try:
        async with semaphore:  
            image = await get_flag(client, base_url, cc)
    except httpx.HTTPStatusError as exc:  
        res = exc.response
        if res.status_code == HTTPStatus.NOT_FOUND:
            status = DownloadStatus.NOT_FOUND
            msg = f'not found: {res.url}'
        else:
            raise
    else:
        await asyncio.to_thread(save_flag, image, f'{cc}.gif')  
        status = DownloadStatus.OK
        msg = 'OK'
    if verbose and msg:
        print(cc, msg)
    return status

async def supervisor(cc_list: list[str],
                     base_url: str,
                     verbose: bool,
                     concur_req: int) -> Counter[DownloadStatus]:  
    counter: Counter[DownloadStatus] = Counter()
    semaphore = asyncio.Semaphore(concur_req)  
    async with httpx.AsyncClient() as client:
        to_do = [download_one(client, cc, base_url, semaphore, verbose)
                 for cc in sorted(cc_list)]  
        to_do_iter = asyncio.as_completed(to_do)  
        if not verbose:
            to_do_iter = tqdm.tqdm(to_do_iter, total=len(cc_list))  
        error: httpx.HTTPError | None = None  
        for coro in to_do_iter:  
            try:
                status = await coro  
            except httpx.HTTPStatusError as exc:
                error_msg = 'HTTP error {resp.status_code} - {resp.reason_phrase}'
                error_msg = error_msg.format(resp=exc.response)
                error = exc  
            except httpx.RequestError as exc:
                error_msg = f'{exc} {type(exc)}'.strip()
                error = exc  
            except KeyboardInterrupt:
                break

            if error:
                status = DownloadStatus.ERROR  
                if verbose:
                    url = str(error.request.url)  
                    cc = Path(url).stem.upper()   
                    print(f'{cc} error: {error_msg}')
            counter[status] += 1

    return counter

def download_many(cc_list: list[str],
                  base_url: str,
                  verbose: bool,
                  concur_req: int) -> Counter[DownloadStatus]:
    coro = supervisor(cc_list, base_url, verbose, concur_req)
    counts = asyncio.run(coro)  

    return counts

if __name__ == '__main__':
    main(download_many, DEFAULT_CONCUR_REQ, MAX_CONCUR_REQ)

async def get_country(client: httpx.AsyncClient,
                      base_url: str,
                      cc: str) -> str:    
    url = f'{base_url}/{cc}/metadata.json'.lower()
    resp = await client.get(url, timeout=3.1, follow_redirects=True)
    resp.raise_for_status()
    metadata = resp.json()  
    return metadata['country']  

async def download_one(client: httpx.AsyncClient,
                       cc: str,
                       base_url: str,
                       semaphore: asyncio.Semaphore,
                       verbose: bool) -> DownloadStatus:
    try:
        async with semaphore:  
            image = await get_flag(client, base_url, cc)
        async with semaphore:  
            country = await get_country(client, base_url, cc)
    except httpx.HTTPStatusError as exc:
        res = exc.response
        if res.status_code == HTTPStatus.NOT_FOUND:
            status = DownloadStatus.NOT_FOUND
            msg = f'not found: {res.url}'
        else:
            raise
    else:
        filename = country.replace(' ', '_')  
        await asyncio.to_thread(save_flag, image, f'{filename}.gif')
        status = DownloadStatus.OK
        msg = 'OK'
    if verbose and msg:
        print(cc, msg)
    return status

One important advantage of Node.js over Python for asynchronous programming is the Node.js standard library, which provides async APIs for all I/O—not just for network I/O. In Python, if you’re not careful, file I/O can seriously degrade the performance of asynchronous applications, because reading and writing to storage in the main thread blocks the event loop.

In the download_one coroutine of Example 21-6, I used this line to save the downloaded image to disk:

        await asyncio.to_thread(save_flag, image, f'{cc}.gif')

As mentioned before, the asyncio.to_thread was added in Python 3.9. If you need to support 3.7 or 3.8, then replace that single line with the lines in Example 21-10.

        loop = asyncio.get_running_loop()         
        loop.run_in_executor(None, save_flag,     
                             image, f'{cc}.gif')  

Get a reference to the event loop.

The first argument is the executor to use; passing None selects the default ThreadPoolExecutor that is always available in the asyncio event loop.

You can pass positional arguments to the function to run, but if you need to pass keyword arguments, then you need to resort to functool.partial, as described in the run_in_executor documentation.

The newer asyncio.to_thread function is easier to use and more flexible, as it also accepts keyword arguments.

The implementation of asyncio itself uses run_in_executor under the hood in a few places. For example, the loop.getaddrinfo(…) coroutine we saw in Example 21-1 is implemented by calling the getaddrinfo function from the socket module—which is a blocking function that may take seconds to return, as it depends on DNS resolution.

A common pattern in asynchronous APIs is to wrap blocking calls that are implementation details in coroutines using run_in_executor internally. That way, you provide a consistent interface of coroutines to be driven with await, and hide the threads you need to use for pragmatic reasons. The Motor asynchronous driver for MongoDB has an API compatible with async/await that is really a façade around a threaded core that talks to the database server. A. Jesse Jiryu Davis, the lead developer of Motor, explains his reasoning in “Response to ‘Asynchronous Python and Databases’”. Spoiler: Davis discovered that a thread pool was more performant in the particular use case of a database driver—despite the myth that asynchronous approaches are always faster than threads for network I/O.

The main reason to pass an explict Executor to loop.run_in_executor is to employ a ProcessPoolExecutor if the function to execute is CPU intensive, so that it runs in a different Python process, avoiding contention for the GIL. Because of the high start-up cost, it would be better to start the ProcessPoolExecutor in the supervisor, and pass it to the coroutines that need to use it.

Caleb Hattingh—the author of Using Asyncio in Python (O’ Reilly)—is one of the tech reviewers of this book and suggested I add the following warning about executors and asyncio.

We’ll now go from client scripts to writing servers with asyncio.

The classic toy example of a TCP server is an echo server. We’ll build slightly more interesting toys: server-side Unicode character search utilities, first using HTTP with FastAPI, then using plain TCP with asyncio only.

These servers let users query for Unicode characters based on words in their standard names from the unicodedata module we discussed in “The Unicode Database”. Figure 21-2 shows a session with web_mojifinder.py, the first server we’ll build.

Figure 21-2. Browser window displaying search results for “mountain” from the web_mojifinder.py service.

The Unicode search logic in these examples is in the InvertedIndex class in the charindex.py module in the Fluent Python code repository. There’s nothing concurrent in that small module, so I’ll only give a brief overview in the optional box that follows. You can skip to the HTTP server implementation in “A FastAPI Web Service”.

A FastAPI Web Service

I wrote the next example—web_mojifinder.py—using FastAPI: one of the Python ASGI Web frameworks mentioned in “ASGI—Asynchronous Server Gateway Interface”. Figure 21-2 is a screenshot of the frontend. It’s a super simple SPA (Single Page Application): after the initial HTML download, the UI is updated by client-side JavaScript communicating with the server.

FastAPI is designed to implement backends for SPA and mobile apps, which mostly consist of web API end points returning JSON responses instead of server-rendered HTML. FastAPI leverages decorators, type hints, and code introspection to eliminate a lot of the boilerplate code for web APIs, and also automatically publishes interactive OpenAPI—a.k.a. Swagger—documentation for the APIs we create. Figure 21-4 shows the autogenerated /docs page for web_mojifinder.py.

Figure 21-4. Autogenerated OpenAPI schema for the /search endpoint.

Example 21-11 is the code for web_mojifinder.py, but that’s just the backend code. When you hit the root URL /, the server sends the form.html file, which has 81 lines of code, including 54 lines of JavaScript to communicate with the server and fill a table with the results. If you’re interested in reading plain framework-less JavaScript, please find 21-async/mojifinder/static/form.html in the Fluent Python code repository.

To run web_mojifinder.py, you need to install two packages and their dependencies: FastAPI and uvicorn.¹⁰ This is the command to run Example 21-11 with uvicorn in development mode:

$ uvicorn web_mojifinder:app --reload

The parameters are:

web_mojifinder:app

The package name, a colon, and the name of the ASGI application defined in it—app is the conventional name.

--reload

Make uvicorn monitor changes to application source files and automatically reload them. Useful only during development.

Now let’s study the source code for web_mojifinder.py.

Example 21-11. web_mojifinder.py: complete source

from pathlib import Path
from unicodedata import name

from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from pydantic import BaseModel

from charindex import InvertedIndex

STATIC_PATH = Path(__file__).parent.absolute() / 'static'  

app = FastAPI(  
    title='Mojifinder Web',
    description='Search for Unicode characters by name.',
)

class CharName(BaseModel):  
    char: str
    name: str

def init(app):  
    app.state.index = InvertedIndex()
    app.state.form = (STATIC_PATH / 'form.html').read_text()

init(app)  

@app.get('/search', response_model=list[CharName])  
async def search(q: str):  
    chars = sorted(app.state.index.search(q))
    return ({'char': c, 'name': name(c)} for c in chars)  

@app.get('/', response_class=HTMLResponse, include_in_schema=False)
def form():  
    return app.state.form

# no main funcion  

Unrelated to the theme of this chapter, but worth noting: the elegant use of the overloaded / operator by pathlib.¹¹

This line defines the ASGI app. It could be as simple as app = FastAPI(). The parameters shown are metadata for the autogenerated documentation.

A pydantic schema for a JSON response with char and name fields.¹²

Build the index and load the static HTML form, attaching both to the app.state for later use.

Run init when this module is loaded by the ASGI server.

Route for the /search endpoint; response_model uses that CharName pydantic model to describe the response format.

FastAPI assumes that any parameters that appear in the function or coroutine signature that are not in the route path will be passed in the HTTP query string, e.g., /search?q=cat. Since q has no default, FastAPI will return a 422 (Unprocessable Entity) status if q is missing from the query string.

Returning an iterable of dicts compatible with the response_model schema allows FastAPI to build the JSON response according to the response_model in the @app.get decorator.

Regular functions (i.e., non-async) can also be used to produce responses.

This module has no main function. It is loaded and driven by the ASGI server—uvicorn in this example.

Example 21-11 has no direct calls to asyncio. FastAPI is built on the Starlette ASGI toolkit, which in turn uses asyncio.

Also note that the body of search doesn’t use await, async with, or async for, therefore it could be a plain function. I defined search as a coroutine just to show that FastAPI knows how to handle it. In a real app, most endpoints will query databases or hit other remote servers, so it is a critical advantage of FastAPI—and ASGI frameworks in general—to support coroutines that can take advantage of asynchronous libraries for network I/O.

Tip

The init and form functions I wrote to load and serve the static HTML form are a hack to make the example short and easy to run. The recommended best practice is to have a proxy/load-balancer in front of the ASGI server to handle all static assets, and also use a CDN (Content Delivery Network) when possible. One such proxy/load-balancer is Traefik, a self-described “edge router” that “receives requests on behalf of your system and finds out which components are responsible for handling them.” FastAPI has project generation scripts that prepare your code to do that.

The typing enthusiast may have noticed that there are no return type hints in search and form. Instead, FastAPI relies on the response_model= keyword argument in the route decorators. The “Response Model” page in the FastAPI documentation explains:

The response model is declared in this parameter instead of as a function return type annotation, because the path function may not actually return that response model but rather return a dict, database object or some other model, and then use the response_model to perform the field limiting and serialization.

For example, in search, I returned a generator of dict items, not a list of CharName objects, but that’s good enough for FastAPI and pydantic to validate my data and build the appropriate JSON response compatible with response_model=list[CharName].

We’ll now focus on the tcp_mojifinder.py script that is answering the queries in Figure 21-5.

An asyncio TCP Server

The tcp_mojifinder.py program uses plain TCP to communicate with a client like Telnet or Netcat, so I could write it using asyncio without external dependencies—and without reinventing HTTP. Figure 21-5 shows text-based UI.

Figure 21-5. Telnet session with the tcp_mojifinder.py server: querying for “fire.”

This program is twice as long as web_mojifinder.py, so I split the presentation into three parts: Example 21-12, Example 21-14, and Example 21-15. The top of tcp_mojifinder.py—including the import statements—is in Example 21-14, but I will start by describing the supervisor coroutine and the main function that drives the program.

Example 21-12. tcp_mojifinder.py: a simple TCP server; continues in Example 21-14

async def supervisor(index: InvertedIndex, host: str, port: int) -> None:
    server = await asyncio.start_server(    
        functools.partial(finder, index),   
        host, port)                         

    socket_list = cast(tuple[TransportSocket, ...], server.sockets)  
    addr = socket_list[0].getsockname()
    print(f'Serving on {addr}. Hit CTRL-C to stop.')  
    await server.serve_forever()  

def main(host: str = '127.0.0.1', port_arg: str = '2323'):
    port = int(port_arg)
    print('Building index.')
    index = InvertedIndex()                         
    try:
        asyncio.run(supervisor(index, host, port))  
    except KeyboardInterrupt:                       
        print('\nServer shut down.')

if __name__ == '__main__':
    main(*sys.argv[1:])

This await quickly gets an instance of asyncio.Server, a TCP socket server. By default, start_server creates and starts the server, so it’s ready to receive connections.

The first argument to start_server is client_connected_cb, a callback to run when a new client connection starts. The callback can be a function or a coroutine, but it must accept exactly two arguments: an asyncio.StreamReader and an asyncio.StreamWriter. However, my finder coroutine also needs to get an index, so I used functools.partial to bind that parameter and obtain a callable that takes the reader and writer. Adapting user functions to callback APIs is the most common use case for functools.partial.

host and port are the second and third arguments to start_server. See the full signature in the asyncio documentation.

This cast is needed because typeshed has an outdated type hint for the sockets property of the Server class—as of May 2021. See Issue #5535 on typeshed.¹³

Display the address and port of the first socket of the server.

Although start_server already started the server as a concurrent task, I need to await on the server_forever method so that my supervisor is suspended here. Without this line, supervisor would return immediately, ending the loop started with asyncio.run(supervisor(…)), and exiting the program. The documentation for Server.serve_forever says: “This method can be called if the server is already accepting connections.”

Build the inverted index.¹⁴

Start the event loop running supervisor.

Catch the KeyboardInterrupt to avoid a distracting traceback when I stop the server with Ctrl-C on the terminal running it.

You may find it easier to understand how control flows in tcp_mojifinder.py if you study the output it generates on the server console, listed in Example 21-13.

Example 21-13. tcp_mojifinder.py: this is the server side of the session depicted in Figure 21-5

$ python3 tcp_mojifinder.py
Building index.  
Serving on ('127.0.0.1', 2323). Hit Ctrl-C to stop.  
 From ('127.0.0.1', 58192): 'cat face'   
   To ('127.0.0.1', 58192): 10 results.
 From ('127.0.0.1', 58192): 'fire'       
   To ('127.0.0.1', 58192): 11 results.
 From ('127.0.0.1', 58192): '\x00'       
Close ('127.0.0.1', 58192).              
^C  
Server shut down.  
$

Output by main. Before the next line appears, I see a 0.6s delay on my machine while the index is built.

Output by supervisor.

First iteration of a while loop in finder. The TCP/IP stack assigned port 58192 to my Telnet client. If you connect several clients to the server, you’ll see their various ports in the output.

Second iteration of the while loop in finder.

I hit Ctrl-C on the client terminal; the while loop in finder exits.

The finder coroutine displays this message then exits. Meanwhile the server is still running, ready to service another client.

I hit Ctrl-C on the server terminal; server.serve_forever is cancelled, ending supervisor and the event loop.

Output by main.

After main builds the index and starts the event loop, supervisor quickly displays the Serving on… message and is suspended at the await server.serve_forever() line. At that point, control flows into the event loop and stays there, occasionally coming back to the finder coroutine, which yields control back to the event loop whenever it needs to wait for the network to send or receive data.

While the event loop is alive, a new instance of the finder coroutine will be started for each client that connects to the server. In this way, many clients can be handled concurrently by this simple server. This continues until a KeyboardInterrupt occurs on the server or its process is killed by the OS.

Now let’s see the top of tcp_mojifinder.py, with the finder coroutine.

Example 21-14. tcp_mojifinder.py: continued from Example 21-12

import asyncio
import functools
import sys
from asyncio.trsock import TransportSocket
from typing import cast

from charindex import InvertedIndex, format_results  

CRLF = b'\r\n'
PROMPT = b'?> '

async def finder(index: InvertedIndex,          
                 reader: asyncio.StreamReader,
                 writer: asyncio.StreamWriter) -> None:
    client = writer.get_extra_info('peername')  
    while True:  
        writer.write(PROMPT)  # can't await!  
        await writer.drain()  # must await!  
        data = await reader.readline()  
        if not data:  
            break
        try:
            query = data.decode().strip()  
        except UnicodeDecodeError:  
            query = '\x00'
        print(f' From {client}: {query!r}')  
        if query:
            if ord(query[:1]) < 32:  
                break
            results = await search(query, index, writer)  
            print(f'   To {client}: {results} results.')  

    writer.close()  
    await writer.wait_closed()  
    print(f'Close {client}.')  

format_results is useful to display the results of InvertedIndex.search in a text-based UI such as the command line or a Telnet session.

To pass finder to asyncio.start_server, I wrapped it with functools.partial, because the server expects a coroutine or function that takes only the reader and writer arguments.

Get the remote client address to which the socket is connected.

This loop handles a dialog that lasts until a control character is received from the client.

The StreamWriter.write method is not a coroutine, just a plain function; this line sends the ?> prompt.

StreamWriter.drain flushes the writer buffer; it is a coroutine, so it must be driven with await.

StreamWriter.readline is a coroutine that returns bytes.

If no bytes were received, the client closed the connection, so exit the loop.

Decode the bytes to str, using the default UTF-8 encoding.

A UnicodeDecodeError may happen when the user hits Ctrl-C and the Telnet client sends control bytes; if that happens, replace the query with a null character, for simplicity.

Log the query to the server console.

Exit the loop if a control or null character was received.

Do the actual search; code is presented next.

Log the response to the server console.

Close the StreamWriter.

Wait for the StreamWriter to close. This is recommended in the .close() method documentation.

Log the end of this client’s session to the server console.

The last piece of this example is the search coroutine, shown in Example 21-15.

Example 21-15. tcp_mojifinder.py: search coroutine

async def search(query: str,  
                 index: InvertedIndex,
                 writer: asyncio.StreamWriter) -> int:
    chars = index.search(query)  
    lines = (line.encode() + CRLF for line  
                in format_results(chars))
    writer.writelines(lines)  
    await writer.drain()      
    status_line = f'{"─" * 66} {len(chars)} found'  
    writer.write(status_line.encode() + CRLF)
    await writer.drain()
    return len(chars)

search must be a coroutine because it writes to a StreamWriter and must use its .drain() coroutine method.

Query the inverted index.

This generator expression will yield byte strings encoded in UTF-8 with the Unicode codepoint, the actual character, its name, and a CRLF sequence—e.g., b'U+0039\t9\tDIGIT NINE\r\n').

Send the lines. Surprisingly, writer.writelines is not a coroutine.

But writer.drain() is a coroutine. Don’t forget the await!

Build a status line, then send it.

Note that all network I/O in tcp_mojifinder.py is in bytes; we need to decode the bytes received from the network, and encode strings before sending them out. In Python 3, the default encoding is UTF-8, and that’s what I used implicitly in all encode and decode calls in this example.

Warning

Note that some of the I/O methods are coroutines and must be driven with await, while others are simple functions. For example, StreamWriter.write is a plain function, because it writes to a buffer. On the other hand, StreamWriter.drain—which flushes the buffer and performs the network I/O—is a coroutine, as is StreamReader.readline—but not StreamWriter.writelines! While I was writing the first edition of this book, the asyncio API docs were improved by clearly labeling coroutines as such.

The tcp_mojifinder.py code leverages the high-level asyncio Streams API that provides a ready-to-use server so you only need to implement a handler function, which can be a plain callback or a coroutine. There is also a lower-level Transports and Protocols API, inspired by the transport and protocols abstractions in the Twisted framework. Refer to the asyncio documentation for more information, including TCP and UDP echo servers and clients implemented with that lower-level API.

Our next topic is async for and the objects that make it work.

We saw in “Asynchronous Context Managers” how async with works with objects implementing the __aenter__ and __aexit__ methods returning awaitables—usually in the form of coroutine objects.

Similarly, async for works with asynchronous iterables: objects that implement __aiter__. However, __aiter__ must be a regular method—not a coroutine method—and it must return an asynchronous iterator.

An asynchronous iterator provides an __anext__ coroutine method that returns an awaitable—often a coroutine object. They are also expected to implement __aiter__, which usually returns self. This mirrors the important distinction of iterables and iterators we discussed in “Don’t Make the Iterable an Iterator for Itself”.

The aiopg asynchronous PostgreSQL driver documentation has an example that illustrates the use of async for to iterate over the rows of a database cursor:

async def go():
    pool = await aiopg.create_pool(dsn)
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            await cur.execute("SELECT 1")
            ret = []
            async for row in cur:
                ret.append(row)
            assert ret == [(1,)]

In this example the query will return a single row, but in a realistic scenario you may have thousands of rows in response to a SELECT query. For large responses, the cursor will not be loaded with all the rows in a single batch. Therefore it is important that async for row in cur: does not block the event loop while the cursor may be waiting for additional rows. By implementing the cursor as an asynchronous iterator, aiopg may yield to the event loop at each __anext__ call, and resume later when more rows arrive from PostgreSQL.

$ python -m asyncio

asyncio REPL 3.9.1 (v3.9.1:1e5d33e9b9, Dec  7 2020, 12:10:52)
[Clang 6.0 (clang-600.0.57)] on darwin
Use "await" directly instead of "asyncio.run()".
Type "help", "copyright", "credits" or "license" for more information.
>>> import asyncio
>>>

>>> await asyncio.sleep(3, 'Rise and shine!')  
'Rise and shine!'
>>> from domainlib import *
>>> await probe('python.org')  
Result(domain='python.org', found=True)  
>>> names = 'python.org rust-lang.org golang.org no-lang.invalid'.split()  
>>> async for result in multi_probe(names):  
...      print(*result, sep='\t')
...
golang.org      True    
no-lang.invalid False
python.org      True
rust-lang.org   True
>>>

>>> probe('python.org')  
<coroutine object probe at 0x10e313740>
>>> multi_probe(names)  
<async_generator object multi_probe at 0x10e246b80>
>>> for r in multi_probe(names):  
...    print(r)
...
Traceback (most recent call last):
   ...
TypeError: 'async_generator' object is not iterable

import asyncio
import socket
from collections.abc import Iterable, AsyncIterator
from typing import NamedTuple, Optional


class Result(NamedTuple):  
    domain: str
    found: bool


OptionalLoop = Optional[asyncio.AbstractEventLoop]  


async def probe(domain: str, loop: OptionalLoop = None) -> Result:  
    if loop is None:
        loop = asyncio.get_running_loop()
    try:
        await loop.getaddrinfo(domain, None)
    except socket.gaierror:
        return Result(domain, False)
    return Result(domain, True)


async def multi_probe(domains: Iterable[str]) -> AsyncIterator[Result]:  
    loop = asyncio.get_running_loop()
    coros = [probe(domain, loop) for domain in domains]  
    for coro in asyncio.as_completed(coros):  
        result = await coro  
        yield result  

    for coro in asyncio.as_completed(coros):
        yield await coro

$ ./domaincheck.py net
FOUND           NOT FOUND
=====           =========
in.net
del.net
true.net
for.net
is.net
                none.net
try.net
                from.net
and.net
or.net
else.net
with.net
if.net
as.net
                elif.net
                pass.net
                not.net
                def.net

#!/usr/bin/env python3
import asyncio
import sys
from keyword import kwlist

from domainlib import multi_probe


async def main(tld: str) -> None:
    tld = tld.strip('.')
    names = (kw for kw in kwlist if len(kw) <= 4)  
    domains = (f'{name}.{tld}'.lower() for name in names)  
    print('FOUND\t\tNOT FOUND')  
    print('=====\t\t=========')
    async for domain, found in multi_probe(domains):  
        indent = '' if found else '\t\t'  
        print(f'{indent}{domain}')


if __name__ == '__main__':
    if len(sys.argv) == 2:
        asyncio.run(main(sys.argv[1]))  
    else:
        print('Please provide a TLD.', f'Example: {sys.argv[0]} COM.BR')

from contextlib import asynccontextmanager

@asynccontextmanager
async def web_page(url):  
    loop = asyncio.get_running_loop()   
    data = await loop.run_in_executor(  
        None, download_webpage, url)
    yield data                          
    await loop.run_in_executor(None, update_stats, url)  

async with web_page('google.com') as data:  
    process(data)

>>> from domainlib import multi_probe
>>> names = 'python.org rust-lang.org golang.org no-lang.invalid'.split()
>>> gen_found = (name async for name, found in multi_probe(names) if found)  
>>> gen_found
<async_generator object <genexpr> at 0x10a8f9700>  
>>> async for name in gen_found:  
...     print(name)
...
golang.org
python.org
rust-lang.org

result = []
async for i in aiter():
    if i % 2:
        result.append(i)

result = [i async for i in aiter() if i % 2]

result = [await fun() for fun in funcs]

>>> names = 'python.org rust-lang.org golang.org no-lang.invalid'.split()
>>> names = sorted(names)
>>> coros = [probe(name) for name in names]
>>> await asyncio.gather(*coros)
[Result(domain='golang.org', found=True),
Result(domain='no-lang.invalid', found=False),
Result(domain='python.org', found=True),
Result(domain='rust-lang.org', found=True)]
>>> [await probe(name) for name in names]
[Result(domain='golang.org', found=True),
Result(domain='no-lang.invalid', found=False),
Result(domain='python.org', found=True),
Result(domain='rust-lang.org', found=True)]
>>>

>>> {name: found async for name, found in multi_probe(names)}
{'golang.org': True, 'python.org': True, 'no-lang.invalid': False,
'rust-lang.org': True}

>>> {name for name in names if (await probe(name)).found}
{'rust-lang.org', 'python.org', 'golang.org'}

Python’s async/await language constructs are not tied to any specific event loop or library.17 Thanks to the extensible API provided by special methods, anyone sufficiently motivated can write their own asynchronous runtime environment and framework to drive native coroutines, asynchronous generators, etc.

That’s what David Beazley did in his Curio project. He was interested in rethinking how these new language features could be used in a framework built from scratch. Recall that asyncio was released in Python 3.4, and it used yield from instead of await, so its API could not leverage asynchronous context managers, asynchronous iterators, and everything else that the async/await keywords made possible. As a result, Curio has a cleaner API and a simpler implementation, compared to asyncio.

Example 21-21 shows the blogdom.py script (Example 21-1) rewritten to use Curio.

#!/usr/bin/env python3
from curio import run, TaskGroup
import curio.socket as socket
from keyword import kwlist

MAX_KEYWORD_LEN = 4


async def probe(domain: str) -> tuple[str, bool]:  
    try:
        await socket.getaddrinfo(domain, None)  
    except socket.gaierror:
        return (domain, False)
    return (domain, True)

async def main() -> None:
    names = (kw for kw in kwlist if len(kw) <= MAX_KEYWORD_LEN)
    domains = (f'{name}.dev'.lower() for name in names)
    async with TaskGroup() as group:  
        for domain in domains:
            await group.spawn(probe, domain)  
        async for task in group:  
            domain, found = task.result
            mark = '+' if found else ' '
            print(f'{mark} {domain}')

if __name__ == '__main__':
    run(main())  

probe doesn’t need to get the event loop, because…

…getaddrinfo is a top-level function of curio.socket, not a method of a loop object—as it is in asyncio.

A TaskGroup is a core concept in Curio, to monitor and control several coroutines, and to make sure they are all executed and cleaned up.

TaskGroup.spawn is how you start a coroutine, managed by a specific TaskGroup instance. The coroutine is wrapped by a Task.

Iterating with async for over a TaskGroup yields Task instances as each is completed. This corresponds to the line in Example 21-1 using for … as_completed(…):.

Curio pioneered this sensible way to start an asynchronous program in Python.

To expand on the last point: if you look at the asyncio code examples for the first edition of Fluent Python, you’ll see lines like these, repeated over and over:

    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

A Curio TaskGroup is an asynchronous context manager that replaces several ad hoc APIs and coding patterns in asyncio. We just saw how iterating over a TaskGroup makes the asyncio.as_completed(…) function unnecessary. Another example: instead of a special gather function, this snippet from the “Task Groups” docs collects the results of all tasks in the group:

async with TaskGroup(wait=all) as g:
    await g.spawn(coro1)
    await g.spawn(coro2)
    await g.spawn(coro3)
print('Results:', g.results)

Task groups support structured concurrency: a form of concurrent programming that constrains all the activity of a group of asynchronous tasks to a single entry and exit point. This is analogous to structured programming, which eschewed the GOTO command and introduced block statements to limit the entry and exit points of loops and subroutines. When used as an asynchronous context manager, a TaskGroup ensures that all tasks spawned inside are completed or cancelled, and any exceptions raised, upon exiting the enclosed block.

Another important feature of Curio is better support for programming with coroutines and threads in the same codebase—a necessity in most nontrivial asynchronous programs. Starting a thread with await spawn_thread(func, …) returns an AsyncThread object with a Task-like interface. Threads can call coroutines thanks to a special AWAIT(coro) function—named in all caps because await is now a keyword.

Curio also provides a UniversalQueue that can be used to coordinate the work among threads, Curio coroutines, and asyncio coroutines. That’s right, Curio has features that allow it to run in a thread along with asyncio in another thread, in the same process, communicating via UniversalQueue and UniversalEvent. The API for these “universal” classes is the same inside and outside of coroutines, but in a coroutine, you need to prefix calls with await.

As I write this in October 2021, HTTPX is the first HTTP client library compatible with Curio, but I don’t know of any asynchronous database libraries that support it yet. In the Curio repository there is an impressive set of network programming examples, including one using WebSocket, and another implementing the RFC 8305—Happy Eyeballs concurrent algorithm for connecting to IPv6 endpoints with fast fallback to IPv4 if needed.

The design of Curio has been influential. The Trio framework started by Nathaniel J. Smith was heavily inspired by Curio. Curio may also have prompted Python contributors to improve the usability of the asyncio API. For example, in its earliest releases, asyncio users very often had to get and pass around a loop object because some essential functions were either loop methods or required a loop argument. In recent versions of Python, direct access to the loop is not needed as often, and in fact several functions that accepted an optional loop are now deprecating that argument.

Type annotations for asynchronous types are our next topic.

The return type of a native coroutine describes what you get when you await on that coroutine, which is the type of the object that appears in the return statements in the body of the native coroutine function.18

This chapter provided many examples of annotated native coroutines, including probe from Example 21-21:

async def probe(domain: str) -> tuple[str, bool]:
    try:
        await socket.getaddrinfo(domain, None)
    except socket.gaierror:
        return (domain, False)
    return (domain, True)

If you need to annotate a parameter that takes a coroutine object, then the generic type is:

class typing.Coroutine(Awaitable[V_co], Generic[T_co, T_contra, V_co]):
    ...

That type, and the following types were introduced in Python 3.5 and 3.6 to annotate asynchronous objects:

class typing.AsyncContextManager(Generic[T_co]):
    ...
class typing.AsyncIterable(Generic[T_co]):
    ...
class typing.AsyncIterator(AsyncIterable[T_co]):
    ...
class typing.AsyncGenerator(AsyncIterator[T_co], Generic[T_co, T_contra]):
    ...
class typing.Awaitable(Generic[T_co]):
    ...

With Python ≥ 3.9, use the collections.abc equivalents of these.

I want to highlight three aspects of those generic types.

First: they are all covariant on the first type parameter, which is the type of the items yielded from these objects. Recall rule #1 of “Variance rules of thumb”:

If a formal type parameter defines a type for data that comes out of the object, it can be covariant.

Second: AsyncGenerator and Coroutine are contravariant on the second to last parameter. That’s the type of the argument of the low-level .send() method that the event loop calls to drive asynchronous generators and coroutines. As such, it is an “input” type. Therefore, it can be contravariant, per Variance Rule of Thumb #2:

If a formal type parameter defines a type for data that goes into the object after its initial construction, it can be contravariant.

Third: AsyncGenerator has no return type, in contrast with typing.Generator, which we saw in “Generic Type Hints for Classic Coroutines”. Returning a value by raising StopIteration(value) was one of the hacks that enabled generators to operate as coroutines and support yield from, as we saw in “Classic Coroutines”. There is no such overlap among the asynchronous objects: AsyncGenerator objects don’t return values, and are completely separate from native coroutine objects, which are annotated with typing.Coroutine.

Finally, let’s briefly discuss the advantages and challenges of asynchronous programming.

How Async Works and How It Doesn’t

The sections closing this chapter discuss high-level ideas around asynchronous programming, regardless of the language or library you are using.

Let’s begin by explaining the #1 reason why asynchronous programming is appealing, followed by a popular myth, and how to deal with it.

Chapter Summary

The problem with normal approaches to asynchronous programming is that they’re all-or-nothing propositions. You rewrite all your code so none of it blocks or you’re just wasting your time.

Alvaro Videla and Jason J. W. Williams, RabbitMQ in Action

I chose that epigraph for this chapter for two reasons. At a high level, it reminds us to avoid blocking the event loop by delegating slow tasks to a different processing unit, from a simple thread all the way to a distributed task queue. At a lower level, it is also a warning: once you write your first async def, your program is inevitably going to have more and more async def, await, async with, and async for. And using non-asynchronous libraries suddenly becomes a challenge.

After the simple spinner examples in Chapter 19, here our main focus was asynchronous programming with native coroutines, starting with the blogdom.py DNS probing example, followed by the concept of awaitables. While reading the source code of flags_asyncio.py, we found the first example of an asynchronous context manager.

The more advanced variations of the flag downloading program introduced two powerful functions: the asyncio.as_completed generator and the loop.run_in_executor coroutine. We also saw the concept and application of a semaphore to limit the number of concurrent downloads—as expected from well-behaved HTTP clients.

Server-side asynchronous programming was presented through the mojifinder examples: a FastAPI web service and tcp_mojifinder.py—the latter using just asyncio and the TCP protocol.

Asynchronous iteration and asynchronous iterables were the next major topic, with sections on async for, Python’s async console, asynchronous generators, asynchronous generator expressions, and asynchronous comprehensions.

The last example in the chapter was blogdom.py rewritten with the Curio framework, to demonstrate how Python’s asynchronous features are not tied to the asyncio package. Curio also showcases the concept of structured concurrency, which may have an industry-wide impact, bringing more clarity to concurrent code.

Finally, the sections under “How Async Works and How It Doesn’t” discuss the main appeal of asynchronous programming, the misconception of “I/O-bound systems,” and dealing with the inevitable CPU-bound parts of your program.

Further Reading

David Beazley’s PyOhio 2016 keynote “Fear and Awaiting in Async” is a fantastic, live-coded introduction to the potential of the language features made possible by Yury Selivanov’s contribution of the async/await keywords in Python 3.5. At one point, Beazley complains that await can’t be used in list comprehensions, but that was fixed by Selivanov in PEP 530—Asynchronous Comprehensions, implemented in Python 3.6 later in that same year. Apart from that, everything else in Beazley’s keynote is timeless, as he demonstrates how the asynchronous objects we saw in this chapter work, without the help of any framework—just a simple run function using .send(None) to drive coroutines. Only at the very end Beazley shows Curio, which he started that year as an experiment to see how far can you go doing asynchronous programming without a foundation of callbacks or futures, just coroutines. As it turns out, you can go very far—as demonstrated by the evolution of Curio and the later creation of Trio by Nathaniel J. Smith. Curio’s documentation has links to more talks by Beazley on the subject.

Besides starting Trio, Nathaniel J. Smith wrote two deep blog posts that I highly recommend: “Some thoughts on asynchronous API design in a post-async/await world”, contrasting the design of Curio with that of asyncio,and “Notes on structured concurrency, or: Go statement considered harmful”, about structured concurrency. Smith also gave a long and informative answer to the question: “What is the core difference between asyncio and trio?” on StackOverflow.

To learn more about the asyncio package, I’ve mentioned the best written resources I know at the start of this chapter: the official documentation after the outstanding overhaul started by Yury Selivanov in 2018, and Caleb Hattingh’s book Using Asyncio in Python (O’Reilly). In the official documentation, make sure to read “Developing with asyncio”: documenting the asyncio debug mode, and also discussing common mistakes and traps and how to avoid them.

For a very accessible, 30-minute introduction to asynchronous programming in general and also asyncio, watch Miguel Grinberg’s “Asynchronous Python for the Complete Beginner”, presented at PyCon 2017. Another great introduction is “Demystifying Python’s Async and Await Keywords”, presented by Michael Kennedy, where among other things I learned about the unsync library that provides a decorator to delegate the execution of coroutines, I/O-bound functions, and CPU-bound functions to asyncio, threading, or multiprocessing as needed.

At EuroPython 2019, Lynn Root—a global leader of PyLadies—presented the excellent “Advanced asyncio: Solving Real-world Production Problems”, informed by her experience using Python as a staff engineer at Spotify.

In 2020, Łukasz Langa recorded a series of great videos about asyncio, starting with “Learn Python’s AsyncIO #1—The Async Ecosystem”. Langa also made the super cool video “AsyncIO + Music” for PyCon 2020 that not only shows asyncio applied in a very concrete event-oriented domain, but also explains it from the ground up.

Another area dominated by event-oriented programming is embedded systems. That’s why Damien George added support for async/await in his MicroPython interpreter for microcontrollers. At PyCon Australia 2018, Matt Trentini demonstrated the uasyncio library, a subset of asyncio that is part of MicroPython’s standard library.

For higher-level thinking about async programming in Python, read the blog post “Python async frameworks—Beyond developer tribalism” by Tom Christie.

Finally, I recommend “What Color Is Your Function?” by Bob Nystrom, discussing the incompatible execution models of plain functions versus async functions—a.k.a. coroutines—in JavaScript, Python, C#, and other languages. Spoiler alert: Nystrom’s conclusion is that the language that got this right is Go, where all functions are the same color. I like that about Go. But I also think Nathaniel J. Smith has a point when he wrote “Go statement considered harmful”. Nothing is perfect, and concurrent programming is always hard.