Making our code nonblocking

It may sound minor, but this is sufficient reason to decide against using this method of making HTTP requests in production code. Without a library to simplify the API here, the cognitive load that is added by using nonblocking sockets is, in my opinion, excessive. The ideal approach would introduce no changes to the program flow, but minimizing the changes helps keep code maintainable. The fact that this implementation leaks the raw sockets into application functions is unacceptable.

Overall, while this approach does reduce waiting time, it requires us to restructure our code significantly, and it only provides savings in the wait step, not in the parsing stage. Nonblocking IO is an interesting technique, but it is only appropriate for exceptional cases and requires significant alterations to program flow as well as abandoning all common libraries to achieve even the most basic outcomes. I don’t recommend this approach.

Low-level threads

The start() method begins the execution of the thread; the join() method blocks the execution until that thread has completed. The name parameter is mostly useful for debugging performance problems, but it’s a good habit always to set a name if you’re ever creating threads manually.

Threads do not have a return value, so if they need to return a computed value, that can be tricky. One way of passing a value back is by using a mutable object that it can change in place or, if using the subclass method, by setting an attribute on the thread object.

An attribute on the thread object is a good approach when there is a single, simple return type, such as a boolean success value, or the result of a computation. It’s a good fit for when the thread is doing a discrete piece of work.

A mutable object is the best fit when you have multiple threads, each working on a part of a common problem, for example, gathering the sensor data from a set of URLs, with each thread being responsible for one URL. The queue.Queue object is perfect for this purpose.

This low level of threading is not at all user-friendly. As we’ve seen in the preceding exercise, it is possible to write a wrapper code that makes functions work unchanged in a threaded environment. It would also be possible to write a wrapper for the threading.Thread object that automatically wraps the function being called and automatically retrieves the result from an internal queue and returns it to the programmer seamlessly.

Luckily, we don’t have to write such a feature in production code; there’s a helper built-in to the Python standard library: concurrent.futures.ThreadPoolExecutor. The ThreadPoolExecutor manages the number of threads in use, allowing the programmer to limit the number of threads that execute at once.

Here, we see a context manager that defines the period where the pool of threads is active. As no max_threads argument is passed to the executor, Python picks an amount of threads based on the number of CPUs available on the computer running the program.

If result() is called within the with block, execution blocks until the relevant task has completed. When the with block ends, execution blocks until all scheduled tasks have completed, so calls to the result method after the with block ends always return immediately.

Bytecode

In order to understand some of the limits of threading in Python, we need to look behind the curtain of how the interpreter loads and runs code. In this section, Python code may be shown annotated with the underlying bytecode used by the interpreter. This bytecode is an implementation detail and is stored in .pyc files. It encodes the behavior of the program at the lowest level. Interpreting a complex language like Python is not a trivial task, so the Python interpreter caches its interpretation of code as a series of many simple operations.

When people talk about Python, they generally are talking about CPython, the implementation of Python in the C programming language. CPython is the reference implementation, in that it’s intended to be what people refer to when seeing how Python does things. There are other implementations, the most popular of which is PyPy, an implementation of Python that is written in a specially designed, Python-like language rather than C.⁴ Both CPython and PyPy cache their interpretation of Python code as Python bytecode.

Two other implementations of Python are worth mentioning: Jython and IronPython. Both of these also cache their interpretation as bytecode, but crucially they use a different bytecode. Jython uses the same bytecode format as Java, and IronPython uses the same bytecode format as .NET. For this chapter, when we talk about bytecode, we’re talking about Python bytecode, as we’re looking at it in the context of how threads are implemented in CPython.

Generally speaking, you won’t have to worry about bytecode, but an awareness of its role is useful for writing multithreaded code. The samples given in the following were generated using the dis module⁵ in the standard library. The function dis.dis(func) shows the bytecode for a given function, assuming that it’s written in Python, rather than a C extension. For example, the sorted(...) function is implemented in C and therefore has no bytecode to show.

The line num += 1 looks like an atomic operation,⁶ but the bytecode reveals that the underlying interpreter runs four operations to complete it. We don’t care what these four instructions are, just the fact that we cannot trust our intuition on what operations are atomic and which are not.

If we were to run this increment function 100 times in succession, the result stored to num would be 100, which makes logical sense. If this function were to be executed in a pair of threads, there would be no guarantee that the final result would be 100. In this case, the correct result is only found so long as no thread ever executes the LOAD_GLOBAL bytecode step while another is running the LOAD_CONST, IN_PLACE_ADD, or STORE_GLOBAL steps. Python does not guarantee this, so the preceding code is not thread-safe.

There is overhead to starting a thread, and the computer will be running multiple processes at the same time. The two threads could happen to run sequentially, despite having two threads available, or they could both start at the same time, or there could be an offset between the start times. The way the executions can overlap is represented in Figure 7-4.

The GIL

That is somewhat simplified, however. CPython has a feature called the GIL, or Global Interpreter Lock, which is used to make thread-safety easier.⁷ This lock means that only one thread can be executing Python code at once. That’s not enough to solve our problem, however, because the granularity of the GIL is at the level of bytecode, so although no two bytecode instructions execute simultaneously, the interpreter can still switch between each path, causing overlap. As such, Figure 7-5 shows a more accurate representation of how the threads can overlap.

It might appear that the GIL removes the benefit of threading without guaranteeing the correct result, but it’s not as bad as it immediately appears. We will deal with the benefits of this shortly, but first, we should address this negating the advantages of threading. It’s not strictly true that no two bytecode instructions can run at once.

Bytecode instructions are much simpler than lines of Python, allowing the interpreter to reason about what actions it is taking at any given point. It can, therefore, allow multiple threads to execute when it’s safe to do so, such as during a network connection or when waiting for data to be read from a file.

Specifically, not everything the Python interpreter does requires the GIL to be held. It must be held at the start and end of a bytecode instruction, but it can be released internally. Waiting for a socket to have data available for reading is one of the things that can be done without holding the GIL. During a bytecode instruction where an IO operation is happening, the GIL can be released, and the interpreter can simultaneously execute any code that does not require the GIL to be held, so long as it’s in a different thread. Once the IO operation finishes, it must wait to regain the GIL from whichever thread took it, before execution can continue.

In situations like this, where the code never has to wait for an IO function to complete, Python interrupts the threads at set intervals to schedule the others fairly. By default, this is approximately every 0.005 seconds which happens to be a long enough period that our example works as hoped on my computer. If we manually tell the interpreter to switch threads more frequently, using the sys.setswitchinterval(...) function, we start to see failures.

Code for testing thread-safety at different switch intervals

if __name__ == "__main__":

    import concurrent.futures

    import sys

    for si in [0.005, 0.0000005, 0.0000000005]:

        sys.setswitchinterval(si)

        results = []

        for attempt in range(100):

            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:

                for i in range(100):

                    pool.submit(increment)

            results.append(num)

            num = 0

        correct = [a for a in results if a == 100]

        pct = len(correct) / len(results)

        print(f"{pct:.1%} correct at sys.setswitchinterval({si:.10f})")

On my computer, the result of running this is

100.0% correct at sys.setswitchinterval(0.0050000000)

71.0% correct at sys.setswitchinterval(0.0000005000)

84.0% correct at sys.setswitchinterval(0.0000000005)

The default behavior being 100% correct in my test doesn’t mean that it solves the problem. 0.005 is a well-chosen interval that results in a lower chance of errors for most people. The fact that a function happens to work when you test it does not mean that it’s guaranteed to always work on every machine. The trade-off to introducing threads is that you gain relatively simple concurrency but without strong guarantees about shared state.

Locks and deadlocks

By enforcing a rule that bytecode instructions do not overlap, they’re guaranteed to be atomic. There is no risk of two STORE bytecode instructions for the same value happening simultaneously, as no two bytecode instructions can run truly simultaneously. It may be that the implementation of an instruction voluntarily releases the GIL and waits to reobtain it for sections of its implementation, but that is not the same as any two arbitrary instructions happening in parallel. This atomicity is used by Python to build thread-safe types and synchronization tools.

If you need to share state between threads, you must manually protect this state with locks. Locks are objects that allow you to prevent your code from running at the same time as other code that it would interfere with. If two concurrent threads both try to acquire the lock, only one will succeed. Any other threads that try to acquire the lock will be blocked until the first thread has released it. This is possible because the locks are implemented in C code, meaning their execution takes place as one bytecode step. All the work of waiting for a lock to become available and acquiring it is performed in response to a single bytecode instruction, making it atomic.

Code protected by a lock can still be interrupted, but no conflicting code will be run during these interruptions. Threads can still be interrupted while they hold a lock. If the thread that they are interrupted in favor of attempts to take that same lock, it will fail to do so and will pause execution. In an environment with two threads, this means execution would pass straight back to the first function. If more than one thread is active, it may pass to other threads first, but the same inability to take the first thread’s lock applies.

In this version of the function, the lock called numlock is used to guard the actions that read/write the num value. This context manager causes the lock to be acquired before execution passes to the body and to be released before the first line after the body. Although we’ve added some overhead here, it is minimal, and it guarantees that the result of the code is correct, regardless of any user settings or different Python interpreter versions.

This code results in the correct result being found no matter what the switching interval is, as the four bytecode instructions that make up num += 1 are guaranteed to be executed as one block. There is an additional locking bytecode instruction before and after each block of four, as shown in Figure 7-6.

From the perspective of the two threads being used in the thread pool, the with numlock : line may block execution, or it may not. Neither thread needs to do anything special to handle the two cases (acquiring the lock immediately or waiting for a turn), and therefore this is a relatively minimal change to the control flow.

The difficulty comes in ensuring that required locks are in place and that no contradictions exist. If a programmer defines two locks and uses them simultaneously, it’s possible to create a situation where the program becomes deadlocked.

Avoiding global state

It’s not always possible to avoid global state, but in many situations, it is. Generally speaking, it’s possible to schedule two functions to run in parallel if neither function depends on the values of shared variables.⁸ Imagine that instead of 100 calls to increment() and 100 calls to decrement(), we were scheduling 100 calls to increment() and 1 to a function called save_number_to_database(). There is no guarantee how many times increment() will have completed before save_number_to_database() is called. The number saved could be anywhere between 0 and 100, which is clearly not useful. These functions don’t make sense to be run in parallel because they both depend on the value of a shared variable.

There are a couple of main ways that shared data can interrelate. Shared data can be used to collate data across multiple threads, or it can be used to pass data between multiple threads.

Passing data

The examples we’ve covered so far all involve the main thread delegating work to subthreads, but it’s common for new tasks to be discovered during the processing of the data from earlier tasks. For example, most APIs paginate data, so if we had a thread to fetch URLs and a thread to parse the responses, we need to be able to pass initial URLs to the fetch thread from the main thread and also to pass newly discovered URLs from the parse thread to the fetch thread.

When passing data between two (or more) threads, we need to use queues, either queue.Queue or the variant queue.LifoQueue. These implement FIFO and LIFO⁹ queues, respectively. While we previously used Queue only as a convenient, thread-safe data holder, now we’ll be using it as intended.

Queues have four primary methods.¹⁰ The get() and put() methods are self-explanatory, except to say that if the queue is empty, then the get() method blocks, and if the queue has a maximum length set and is full, then the put() method blocks. In addition, there is a task_done() method , which is used to tell the queue that an item has been successfully processed, and a join() method, which blocks until all items have been successfully processed. The join() method is usually called by the thread that adds the items to the queue, to allow it to wait until all work has been completed.

Because the get() method blocks if the queue is currently empty, it’s not possible to use this method in nonthreaded code. It does, however, make them perfect for threaded code where there is a need to wait until the thread producing data has made it available.

Tip

It’s not always clear in advance how many items will be stored in a queue. If get() is called after the last item has been retrieved, then it will block indefinitely. This can be avoided by providing a timeout parameter to get, in which case it will block for the given amount of seconds before raising a queue.Empty exception. A better approach is to send a sentinel value, like None. The code can then detect this value and know that it no longer needs to retrieve new values.

If we were building a threaded program to get information from the GitHub public API, we’d need to be able to retrieve URLs and parse their results. It would be nice to be able to do parsing while URLs are being fetched, so we would split the code between fetching and parsing functions.

Listing 7-5 shows an example of such a program, where multiple GitHub repos can have their commits retrieved in parallel. It uses three queues, one for the input to the fetch thread, one for the output of fetch and input of parse, and one for the output of parse.

from concurrent.futures import ThreadPoolExecutor

import queue

import requests

import textwrap

def print_column(text, column):

    wrapped = textwrap.fill(text, 45)

    indent_level = 50 * column

    indented = textwrap.indent(wrapped, " " * indent_level)

    print(indented)

def fetch(urls, responses, parsed):

    while True:

        url = urls.get()

        if url is None:

            print_column("Got instruction to finish", 0)

            return

        print_column(f"Getting {url}", 0)

        response = requests.get(url)

        print_column(f"Storing {response} from {url}", 0)

        responses.put(response)

        urls.task_done()

def parse(urls, responses, parsed):

    # Wait for the initial URLs to be processed

    print_column("Waiting for url fetch thread", 1)

    urls.join()

    while not responses.empty():

        response = responses.get()

        print_column(f"Starting processing of {response}", 1)

        if response.ok:

            data = response.json()

            for commit in data:

                parsed.put(commit)

            links = response.headers["link"].split(",")

            for link in links:

                if "next" in link:

                    url = link.split(";")[0].strip("<>")

                    print_column(f"Discovered new url: {url}", 1)

                    urls.put(url)

        responses.task_done()

        if responses.empty():

            # We have no responses left, so the loop will

            # end. Wait for all queued urls to be fetched

            # before continuing

            print_column("Waiting for url fetch thread", 1)

            urls.join()

    # We reach this point if there are no responses to process

    # after waiting for the fetch thread to catch up. Tell the

    # fetch thread that it can stop now, then exit this thread.

    print_column("Sending instruction to finish", 1)

    urls.put(None)

def get_commit_info(repos):

    urls = queue.Queue()

    responses = queue.Queue()

    parsed = queue.Queue()

    for (username, repo) in repos:

        urls.put(f"https://api.github.com/repos/{username}/{repo}/commits")

    with ThreadPoolExecutor() as pool:

        fetcher = pool.submit(fetch, urls, responses, parsed)

        parser = pool.submit(parse, urls, responses, parsed)

    print(f"{parsed.qsize()} commits found")

if __name__ == "__main__":

    get_commit_info(

        [("MatthewWilkes", "apd.sensors"), ("MatthewWilkes", "apd.aggregation")]

    )

Listing 7-5

Threaded API client

Running this code results in a two-column output, consisting of the messages from each thread. The full output is too long to include here, but a small section is given in the following as a demonstration:

Getting https://api.github.com/repos/MatthewW

ilkes/apd.aggregation/commits

Storing <Response [200]> from https://api.git

hub.com/repos/MatthewWilkes/apd.aggregation/c

ommits

                                                  Starting processing of <Response [200]>

                                                  Discovered new url: https://api.github.com/

                                                  repositories/188280485/commits?page=2

                                                  Starting processing of <Response [200]>

Getting https://api.github.com/repositories/1

88280485/commits?page=2

                                                  Discovered new url: https://api.github.com/

                                                  repositories/222268232/commits?page=2

By examining the logged messages from each of the threads, we can view how their work is scheduled in parallel. Firstly, the main thread sets up the necessary queues and subthreads, then waits for all the threads to finish. As soon as the two subthreads start, the fetch thread starts working on the URLs passed by the main thread, and the parse thread quickly pauses while waiting for responses to parse.

The parse thread uses urls.join() when there is no work for it, so whenever it runs out of work, it waits until the fetch thread has caught up with all the work that it was sent. This is visible in Figure 7-7, as the parse lines always resume after the fetch lines are complete.

The fetch thread doesn’t use the join() method of any of the queues, it uses get() to block until there is some work to do. As such, the fetch thread can be seen resuming while the parse thread is still executing. Finally, the parse thread sends a sentinel value to the fetch thread to end, and when both exits the thread pool context manager in the main thread exits and execution returns to the main thread.

Other synchronization primitives

The synchronization we used with queues in the preceding example is more complex than the lock behavior we used earlier. In fact, there are a variety of other synchronization primitives available in the standard library. These allow you to build more complex thread-safe coordination behaviors.

ProcessPoolExecutors

Just as we’ve looked at the use of ThreadPoolExecutor to delegate the execution of code to different threads, which causes us to fall foul of the GIL’s restrictions, we can use the ProcessPoolExecutor to run code in multiple processes if we’re willing to abandon all shared state.

When executing code in a process pool, any state that was available at the start is available to the subprocesses. However, there is no coordination between the two. Data can only be passed back to the controlling process as the return value of the tasks submitted to the pool. No changes to global variables are reflected in any way.

Although multiple independent Python processes are not bound by the same one-at-a-time execution method imposed by the GIL, they also have significant overheads. For IO-bound tasks (i.e., tasks that spend most of their time waiting and therefore not holding the GIL), a process pool is generally slower than a thread pool.

On the other hand, tasks that involve large amounts of computation are well suited to being delegated to a subprocess, especially long-running ones where the overhead of the setup is lessened compared to the savings of parallel execution.

Making our code multithreaded

The function that we want to parallelize is get_data_points(...); the functions that implement the command line and database connections do not significantly change when dealing with 1 or 500 sensors; there is no particular reason to split its work out into threads. Keeping this work in the main thread makes it easier to handle errors and report on progress, so we rewrite the add_data_from_sensors(...) function only.

As we will submit all our jobs to the ThreadPoolExecutor before the first time that we call a result() method, they will all be queued up for simultaneous execution in threads. It’s the result() method and the end of the with block that triggers blocking; submitting jobs does not cause the program to block, even if you submit more jobs than can be processed simultaneously.

This method is much less intrusive to the program flow than either the raw threaded approach or the nonblocking IO approaches, but it does still involve changing the execution flow to handle the fact that these functions are now working with Future objects rather than the data directly.

async def

This behaves in the same way: two coroutines are run, their values are retrieved, and the num variable is adjusted according to the result of those functions. The main difference is that instead of these coroutines being submitted to a thread pool, the onehundred() async function is passed to the event loop to run, and that function is responsible for calling the other coroutines that do the work.

The asyncio.run(...) function is the main entrypoint for asynchronous code. It blocks until the passed function, and all other functions which that function schedules, are complete. The upshot is that only one coroutine at a time can be initiated by synchronous code.

await

The await keyword is the trigger for blocking until an asynchronous function has completed. However, this only blocks the current asynchronous call stack. You can have multiple asynchronous functions executing at once, in which case another function is executed while waiting for the result.

The await keyword is equivalent to the Future.result() method in the ThreadPoolExecutor example: it transforms an awaitable object into its result. It can appear wherever an asynchronous function call is used; it’s equally valid to write any of the three variants of printing the result of a function shown in Figure 7-8 .

An awaitable object is an object that implements the __await__() method . This is an implementation detail; you won’t need to write an __await__() method. Instead, you will use a variety of different built-in objects that provide it for you. For example, any coroutines defined using async def have an __await__() method.

There are some useful convenience functions to handle scheduling tasks based on coroutines, most notably asyncio.gather(...) . This method takes any number of awaitable objects, schedules them as tasks, awaits them all, and returns an awaitable of a tuple of their return values in the same order that their coroutines/tasks were originally given in.

async for

The async for construct allows iterating over an object where the iterator itself is defined by asynchronous code. It is not correct to use async for on synchronous iterators that are merely being used in an asynchronous context or that happen to contain awaitables.

None of the common data types we’ve used are asynchronous iterators. If you have a tuple or a list, then you use the standard for loop, regardless of what they contain or if they’re being used in synchronous or asynchronous code.

This section contains examples of three different approaches to looping in an asynchronous function. Type hinting is especially useful here, as the data types here are subtly different, and it makes it clear which types each function expects.

In the add_all(...) function, the loop is a standard for loop, as it’s iterating over a list. The contents of the list are the result of number(2) and number(3), so these two calls need to be awaited to retrieve their respective results.

The numbers() coroutine is now responsible for awaiting the individual number(...) coroutines. We still use a standard for loop, but now instead of awaiting the contents of the for loop, we await the value we’re looping over.

With both approaches, the first number(...) call is awaited before the second, but with the first approach, control passes back to the add_all(...) function between the two. In the second, control is only passed back after all numbers have been awaited individually and assembled into a list. With the first method, each number(...) coroutine is processed as it’s needed, but with the second all processing of the number(...) calls happens before the first value is used.

We still need the await keywords in the numbers() method as we want to iterate over the results of the number(...) method, not over placeholders for the results. Like the second version, this hides the details of awaiting the individual number(...) calls from the sum(...) function, instead of trusting the iterator to manage it. However, it also retains the property of the first that each number(...) call is only evaluated when it’s needed: they’re not all processed in advance.

For an object to support being iterated over with for, it must implement an __iter__ method that returns an iterator. An iterator is an object that implements both an __iter__ method (that returns itself) and a __next__ method for advancing the iterator. An object that implements __iter__ but not __next__ is not an iterator but an iterable. Iterables can be iterated over; iterators are also aware of their current state.

Equally, an object that implements an asynchronous method __aiter__ is an AsyncIterable. If __aiter__ returns self and it also provides an __anext__ asynchronous method, it’s an AsyncIterator.

A single object can implement all four methods to support both synchronous and asynchronous iterations. This is only relevant if you’re implementing a class that can behave as an iterable, either synchronous or asynchronous. The easiest way to create an async iterable is using the yield construct from an async function, and that’s enough for most use cases.

async with

The with statement also has an async counterpart, async with, which is used to facilitate the writing of context managers that depend on asynchronous code. It’s quite common to see this in asynchronous code, as many IO operations involve setup and teardown phases.

In the same way that async for uses __aiter__ rather than __iter__, asynchronous context managers define the __aenter__ and __aexit__ methods to replace __enter__ and __exit__. Once again, objects can choose to implement all four to work in both contexts, if appropriate.

When using synchronous context managers in an asynchronous function, there’s a potential for blocking IO to happen before the first line and after the last line of the body. Using async with and a compatible context manager allows for the event loop to schedule some other asynchronous code during that blocking IO period.

We will cover using and creating context managers in more detail over the next two chapters, but both are equivalent to try/finally constructions, but standard context managers use synchronous code in their enter and exit methods, whereas async context managers use asynchronous code.

Async locking primitives

Although asynchronous code is less vulnerable to concurrency safety issues than threads are, it is still possible to write asynchronous code that has concurrency bugs. The switching model being based on awaiting a result rather than threads being interrupted prevents most accidental bugs, but it’s no guarantee of correctness.

Although this program should print 100, it may print any number as low as 1, depending on the decisions the event loop makes about scheduling tasks. To prevent this, we need to either move the await offset() call to not be part of the += construction or lock the num variable.

Perhaps the biggest difference between threaded and async versions of synchronization primitives is that async primitives cannot be defined at the global scope. More accurately, they can only be instantiated from within a running coroutine as they must register themselves with the current event loop.

Working with synchronous libraries

The run_in_executor(...) function works by switching out the problem to one that is easily made asynchronous. Instead of trying to turn arbitrary Python functions from synchronous into asynchronous, finding the right places to yield control back to the event loop, getting woken up at the correct time, and so on, it uses a thread (or a process) to execute the code. Threads and processes are inherently suitable to asynchronous control by virtue of being an operating system construct. This reduces the scope of what needs to be made compatible with the asyncio system to starting a thread and waiting for it to be complete.

Making our code asynchronous

The first step in making our code work in an asynchronous context is to pick a function to act as the first in the chain of asynchronous functions. We want to keep the synchronous and asynchronous code separate, so we need to pick something that’s high enough in the call stack that all things that need to be asynchronous are (perhaps indirectly) called by this function.

In our code, the get_data_points(...) function is the only one that we want to run in an asynchronous context. It is called by add_data_from_sensors(...), which is itself called by standalone(...), which is called by collect_sensor_data(...) in turn. Any of these four functions can be the argument to asyncio.run(...).

The collect_sensor_data(...) function is the click entrypoint, so it cannot be an asynchronous function. The get_data_points(...) function needs to be called multiple times, so it is a better fit for a coroutine than the main entrypoint into the asynchronous flow. This leaves standalone(...) and add_data_from_sensors(...).

We now need to change our implementations of the lower-level functions to not call any blocking synchronous code. Currently, we are making our HTTP calls using the requests library, which is a blocking, synchronous library.

As an alternative, we’ll switch to the aiohttp module to make our HTTP requests. Aiohttp is a natively asynchronous HTTP library that supports both client and server applications. The interface is not as refined as that of requests, but it is quite usable.

As the name suggests, a ClientSession represents the idea of a session with a shared cookie state and HTTP header configuration. Within this, requests are made with asynchronous context managers like get. The result of the context manager is an object which has methods that can be awaited to retrieve the contents of the response.

The preceding construction, which is admittedly much more verbose than the equivalent using requests, allows for many places where the execution flow could be yielded to work around blocking IO. The obvious one is the await line, which relinquishes control while waiting for the response to be retrieved and parsed as JSON. Less obvious is the entry and exit of the http.get(...) context manager, which can set up socket connections, allowing things like DNS resolution not to block execution. It’s also possible for the execution flow to be yielded when entering and exiting a ClientSession.

All this is to say that while the preceding construction is more verbose than the same code using requests, it does allow for transparently setting up and tearing down of shared resources relating to the HTTP session and is doing so in a way that does not significantly slow the process.

In this function, we define two variables, a list of awaitables that each return a list of DataPoint objects, as well as a list of DataPoint objects that we fill as we process the awaitables. Then, we set up the ClientSession and iterate over the servers, adding an invocation of get_data_points(...) for each server. At this stage, these are coroutines as they are not scheduled as a task. We could await them in turn, but this would have the effect of making each request happen sequentially. Instead, we use asyncio.gather(...) to schedule them as tasks and allow us to iterate over the results, which are each a list of DataPoint objects.

Next, we need to add the data to the database. We’re using SQLAlchemy here, which is a synchronous library. For production-quality code, we’d need to ensure that there is no chance of blocking here. The following implementation does not guarantee that the session.add(...) method can block due to the data being synchronized with the database session.

We will look at methods for dealing with database integration in a parallel execution context in the next chapter, but this is good enough for a prototype.

Finally, we need to do the actual work of getting the data. The method is greatly different to the synchronous version, except that it also requires the ClientSession to be passed in, and some minor changes must be made to accommodate the difference in HTTP request API.

This method makes many different choices when compared to a multithreaded or a multiprocess model. A multiprocess model allows for true concurrent processing, and a multithreaded approach can achieve some very minor performance gains thanks to the less restrictive guarantees about switching, but asynchronous code has a much more natural interface, in my opinion.

The key disadvantage of the asyncio approach is that the advantages can only be truly realized with asynchronous libraries. Other libraries can still be used by combining asyncio and threaded approaches, which is made easy by the good integration between these two methods, but there is a significant refactoring requirement to converting existing code to an asynchronous approach and, equally, a significant learning curve in becoming accustomed to writing asynchronous code in the first place.

Additional resources