Async Programming in Python: From Generators to asyncio

Synchronous Programming

This first example shows a somewhat contrived way of having a task retrieve work from a queue and process that work. A queue in Python is a nice FIFO (first in, first out) data structure. It provides methods to put things in a queue and take them out again in the order they were inserted.

In this case, the work is to get a number from the queue and have a loop count up to that number. It prints to the console when the loop begins and again to output the total. This program demonstrates one way for multiple synchronous tasks to process the work in a queue.

The program named example_1.py in the downloadable code is listed in full below:

 1importqueue
 2
 3deftask(name, work_queue):
 4    if work_queue.empty():
 5        print(f"Task {name} nothing to do")
 6        return
 7
 8    while not work_queue.empty():
 9        count = work_queue.get()
10        total = 0
11        print(f"Task {name} running")
12        for _ in range(count):
13            total += 1
14        print(f"Task {name} total: {total}")
15
16defmain():
17    work_queue = queue.Queue()
18    for work in [15, 10, 5, 2]:
19        work_queue.put(work)
20
21    tasks = [(task, "One", work_queue), (task, "Two", work_queue)]
22
23    for task_func, task_name, tasks_queue in tasks:
24        task_func(task_name, tasks_queue)
25
26if __name__ == "__main__":
27    main()

Here’s what each line does:

• Line 1 imports the queue module. This is where the program stores work to be done by the tasks.
• Lines 3 to 14 define task(). This function pulls work out of work_queue and processes the work until there isn’t any more to do.
• Lines 4 to 6 cover the case where the queue is already empty. The task says it has nothing to do and returns right away.
• Line 16 defines main() to run the program tasks.
• Line 17 creates the work_queue. All tasks use this shared resource to retrieve work.
• Lines 18 to 19 put work in work_queue. In this case, it’s just a list of counts for the tasks to process.
• Line 21 creates a list of task tuples, with the parameter values those tasks will be passed.
• Lines 23 to 24 iterate over the list, calling each task and passing the values stored alongside it.
• Line 27 calls main() to run the program.

The task in this program is just a function accepting a string and a queue as parameters. When executed, it looks for anything in the queue to process. If there is work to do, then it pulls values off the queue, starts a for loop to count up to that value, and outputs the total at the end. It continues getting work off the queue until there is nothing left and it exits.

When this program is run, it produces the output you see below:

$ pythonexample_1.py
Task One running
Task One total: 15
Task One running
Task One total: 10
Task One running
Task One total: 5
Task One running
Task One total: 2
Task Two nothing to do

This shows that Task One does all the work. The while loop that Task One hits within task() consumes all the work on the queue and processes it. When that loop exits, Task Two gets a chance to run. However, it finds that the queue is empty, so Task Two prints a statement that says it has nothing to do and then exits. There’s nothing in the code to allow both Task One and Task Two to switch contexts and work together.

Simple Cooperative Concurrency

The next version of the program allows the two tasks to work together. Adding a yield statement means the loop will yield control at the specified point while still maintaining its context. This way, the yielding task can be restarted later.

The yield statement turns task() into a generator function. Calling task() still looks like an ordinary function call, but it doesn’t run the body. You get a generator object back instead, and the body starts only when you call next() on it. From then on, each yield returns control to the caller. This is essentially a context switch, as control moves from the generator function to the caller.

The interesting part is that control can be given back to the generator function by calling next() on the generator. This is a context switch back to the generator function, which picks up execution with all function variables that were defined before the yield still intact.

The while loop in main() takes advantage of this when it calls next(current_task). This statement restarts the task at the point where it previously yielded. All of this means that you’re in control when the context switch happens: when the yield statement is executed in task().

This is a form of cooperative multitasking. The program yields control of its current context so that something else can run. Here, that lets the while loop in main() run two instances of task() as a generator function, with each instance consuming work from the same queue. It’s clever, but it’s also a lot of machinery to get the same results as the first program. The program example_2.py demonstrates this basic concurrency and is listed below:

 1importqueue
 2
 3deftask(name, work_queue):
 4    while not work_queue.empty():
 5        count = work_queue.get()
 6        total = 0
 7        print(f"Task {name} running")
 8        for _ in range(count):
 9            total += 1
10            yield
11        print(f"Task {name} total: {total}")
12
13defmain():
14    work_queue = queue.Queue()
15    for work in [15, 10, 5, 2]:
16        work_queue.put(work)
17
18    tasks = [task("One", work_queue), task("Two", work_queue)]
19
20    while tasks:
21        for current_task in tasks.copy():
22            try:
23                next(current_task)
24            except StopIteration:
25                tasks.remove(current_task)
26
27if __name__ == "__main__":
28    main()

Here’s what’s happening in the code above:

• Lines 3 to 11 define task() much as before, minus the empty-queue guard, but the yield on line 10 turns the function into a generator. This is where the context switch happens, and control passes back to the while loop in main().
• Line 18 creates the task list. Because task() now contains yield, calling it doesn’t run the function body at all. It returns a generator object, which the list holds until you advance it with next().
• Lines 20 to 25 replace the for loop in main() with a while loop so that task() can run cooperatively. Control returns to each instance of task() when it yields, letting the loop continue and run another task.
• Line 20 keeps looping while tasks still holds something. Each exhausted generator gets removed from the list, so the loop ends on its own once both tasks finish.
• Line 21 iterates over a copy of tasks, so that removing an item on line 25 doesn’t disturb the iteration.
• Line 23 gives control back to task(), continuing its execution after the point where yield was called.

This is the output produced when you run this program:

$ pythonexample_2.py
Task One running
Task Two running
Task Two total: 10
Task Two running
Task One total: 15
Task One running
Task Two total: 5
Task One total: 2

You can see that both Task One and Task Two are running and consuming work from the queue. This is what’s intended, as both tasks are processing work, and each is responsible for two items in the queue. This is interesting, but again, it takes quite a bit of work to achieve these results.

The trick here is using the yield statement, which turns task() into a generator and performs a context switch. The program uses this context switch to give control to the while loop in main(), allowing two instances of a task to run cooperatively.

Notice how Task Two outputs its total first. That might lead you to think the tasks are running asynchronously, but this is still a synchronous program, structured so the two tasks can trade contexts back and forth.

Task Two finishes first only because it counts to 10 while Task One counts to 15. It reaches its total sooner, so it prints to the console first.

That back-and-forth is easier to follow when you can watch it happen. In the figure below, three tasks share a single CPU, and each one keeps it until it reaches its next yield:

Nothing here runs at the same time, since the CPU only ever changes hands when the task holding it chooses to give it up.

Real programs spend much of their time waiting on slow things like disk reads and network responses, and that idle time is what async code is built to reclaim. To see the difference, your tasks need something to wait for.

Cooperative Concurrency With Blocking Calls

The next version of the program keeps the same cooperative structure, except that a time.sleep(delay) replaces the counting loop in the body of your task. This adds a delay based on the value retrieved from the work queue to every iteration of the task loop. The delay simulates the effect of a blocking call occurring in your task.

A blocking call is code that stops the CPU from doing anything else for some period of time. In the parent analogy from the first section, if a parent wasn’t able to break away from balancing the checkbook until it was complete, that would be a blocking call.

time.sleep(delay) does the same thing in this example, because the CPU can’t do anything but wait for the delay to expire.

Here’s example_3.py with that blocking delay in place:

 1importqueue
 2importtime
 3
 4fromcodetimingimport Timer
 5
 6deftask(name, work_queue):
 7    timer = Timer(text=f"Task {name} elapsed time: {{:.1f}}")
 8    while not work_queue.empty():
 9        delay = work_queue.get()
10        print(f"Task {name} running")
11        timer.start()
12        time.sleep(delay)
13        timer.stop()
14        yield
15
16defmain():
17    work_queue = queue.Queue()
18    for work in [15, 10, 5, 2]:
19        work_queue.put(work)
20
21    tasks = [task("One", work_queue), task("Two", work_queue)]
22
23    with Timer(text="\nTotal elapsed time: {:.1f}"):
24        while tasks:
25            for current_task in tasks.copy():
26                try:
27                    next(current_task)
28                except StopIteration:
29                    tasks.remove(current_task)
30
31if __name__ == "__main__":
32    main()

Here’s what’s different in the code above:

• Line 2 imports the time module to give the program access to time.sleep().
• Line 4 imports Timer from the codetiming module.
• Line 7 creates the Timer instance that measures how long each iteration of the task loop takes.
• Line 11 starts the timer instance.
• Line 12 adds time.sleep(delay) to task() to mimic an I/O delay. This replaces the for loop that did the counting in example_2.py.
• Line 13 stops the timer instance and outputs the elapsed time since timer.start() was called.
• Line 14 yields control back to main(). In example_2.py, the yield sat inside the counting loop, so it fired on every increment. Now that the counting is gone, it fires once per item pulled off the queue.
• Line 23 creates a Timer context manager that outputs how long the entire while loop took to execute.

When you run this program, you’ll see the following output:

$ pythonexample_3.py
Task One running
Task One elapsed time: 15.0
Task Two running
Task Two elapsed time: 10.0
Task One running
Task One elapsed time: 5.0
Task Two running
Task Two elapsed time: 2.0

Total elapsed time: 32.0

As before, both Task One and Task Two are running, consuming work from the queue and processing it. However, even with the addition of the delay, you can see that cooperative concurrency hasn’t gotten you anything. The delay stops the processing of the entire program, and the CPU just waits for the I/O delay to be over.

This is exactly what the asyncio documentation means by blocking code. You’ll notice that the time it takes to run the entire program is just the cumulative time of all the delays. Running tasks this way is not a win.

Cooperative Concurrency With Non-Blocking Calls

The next version of the program has been modified quite a bit. It makes use of the asyncio package and the await keyword.

The time and queue modules have been replaced with the asyncio package. This gives your program access to asynchronous-friendly (non-blocking) sleep and queue functionality. The async prefix on line 5 turns task() into an async function.

The other big change is removing the time.sleep(delay) and yield statements and replacing them with await asyncio.sleep(delay). This creates a non-blocking delay that will perform a context switch back to the caller main().

The while loop inside main() no longer exists. In its place, an async with asyncio.TaskGroup() block creates both tasks and waits for them. A task group tells asyncio two things:

1. Schedule each coroutine handed to group.create_task() and start running it.
2. Wait for every task in the group to finish before leaving the async with block.

The last line of the program, asyncio.run(main()), runs main(). This creates what’s known as an event loop. It’s this loop that runs main(), which in turn runs the two instances of task().

The event loop is at the heart of the Python async system. It runs all the code, including main(). When task code is executing, the CPU is busy doing work. When the await keyword is reached, a context switch occurs, and control passes back to the event loop. The event loop looks at all the tasks waiting for an event (in this case, an asyncio.sleep(delay) timeout) and passes control to a task with an event that’s ready.

await asyncio.sleep(delay) is non-blocking as far as the CPU is concerned. Instead of waiting for the delay to expire, the CPU registers a sleep event on the event loop’s task queue and hands control back to the loop.

The event loop watches for completed events and passes control back to whichever task was waiting on one. That way the CPU stays busy whenever work is available, while the loop keeps track of events due in the future.

The example_4.py code is listed below:

 1importasyncio
 2
 3fromcodetimingimport Timer
 4
 5async deftask(name, work_queue):
 6    timer = Timer(text=f"Task {name} elapsed time: {{:.1f}}")
 7    while not work_queue.empty():
 8        delay = await work_queue.get()
 9        print(f"Task {name} running")
10        timer.start()
11        await asyncio.sleep(delay)
12        timer.stop()
13
14async defmain():
15    work_queue = asyncio.Queue()
16    for work in [15, 10, 5, 2]:
17        await work_queue.put(work)
18
19    with Timer(text="\nTotal elapsed time: {:.1f}"):
20        async with asyncio.TaskGroup() as group:
21            group.create_task(task("One", work_queue))
22            group.create_task(task("Two", work_queue))
23
24if __name__ == "__main__":
25    asyncio.run(main())

Here’s what’s different between this program and example_3.py:

• Line 1 imports asyncio to gain access to the async machinery. This replaces the time import.
• Line 3 imports Timer from the codetiming module.
• Line 5 adds the async keyword in front of the task() definition, which makes it an async function that can run asynchronously.
• Line 6 creates the Timer instance that measures how long each iteration of the task loop takes.
• Line 10 starts the timer instance.
• Line 11 replaces time.sleep(delay) with the non-blocking asyncio.sleep(delay), which also yields control back to the main event loop.
• Line 12 stops the timer instance and outputs the elapsed time since timer.start() was called.
• Line 15 creates the non-blocking asynchronous work_queue.
• Lines 16 to 17 put work into work_queue asynchronously using the await keyword.
• Line 19 creates a Timer context manager that outputs how long the whole run took.
• Lines 20 to 22 open a task group and schedule the two tasks in it. Leaving the async with block waits for both to finish.
• Line 25 starts the program running asynchronously. It also starts the internal event loop.

When you look at the output of this program, notice how both Task One and Task Two start at the same time, then wait at the mock I/O call:

$ pythonexample_4.py
Task One running
Task Two running
Task Two elapsed time: 10.0
Task Two running
Task One elapsed time: 15.0
Task One running
Task Two elapsed time: 5.0
Task One elapsed time: 2.0

Total elapsed time: 17.0

This indicates that await asyncio.sleep(delay) is non-blocking and that other work is being done.

At the end of the program, you’ll notice the total elapsed time is essentially half the time it took for example_3.py to run. That’s the advantage of a program that uses Python async features! Each task was able to run await asyncio.sleep(delay) at the same time. The total execution time of the program is now less than the sum of its parts. You’ve broken away from the synchronous model!

Both programs spend the same total time waiting. What changes is whether those waits overlap:

Whichever queue you pick, the blocking run costs the sum of every wait, while the async run can be no shorter than its longest single one. Your code still runs one line at a time, so it’s the waiting that overlaps, not the code.

Async Task Inspection

Once a program has more than a couple of tasks in flight, you’ll want to see what the event loop is actually doing. Python 3.14 added a command-line tool for that. It attaches to a running process and reports the tasks the event loop is tracking, without you adding any instrumentation to your code.

Start example_4.py in one terminal. While it’s still running, note its process ID and pass that to python -m asyncio pstree in a second terminal:

$ python-masynciopstree29421
└── (T) Task-1
    └──  main example_4.py:22
        └──  TaskGroup.__aexit__ asyncio/taskgroups.py:72
            └──  TaskGroup._aexit asyncio/taskgroups.py:121
                ├── (T) Task-2
                │   └──  task example_4.py:12
                │       └──  sleep asyncio/tasks.py:704
                └── (T) Task-3
                    └──  task example_4.py:12
                        └──  sleep asyncio/tasks.py:704

The tree makes the shape of your program visible. Task-1 is main() itself, parked in the task group’s __aexit__() while it waits. Hanging off it are the two workers you scheduled in the task group, and each one is suspended at the await asyncio.sleep(delay) inside task().

There’s also a ps subcommand, which reports the same tasks as a flat table. It’s wider than the tree, but it adds explicit awaiter columns that spell out which task is waiting on which. When a program stops making progress, either view is usually the quickest way to find out what’s stuck and what it’s waiting on.

Between the elapsed times your programs print and the task tree you can pull from a running process, you can now see both how long your async code takes and what the event loop is doing while it waits.

Synchronous (Blocking) HTTP Calls

This version returns to the generator-based structure of example_3.py, but with real work in place of the simulated delay. It makes HTTP requests to a list of URLs and gets the page contents, and it does so in a blocking (synchronous) manner. That gives you a baseline to measure the async version against.

The program has been modified to import the wonderful requests module to make the actual HTTP requests. Also, the queue now contains a list of URLs, rather than numbers. In addition, task() no longer increments a counter. Instead, it uses requests to get the contents of a URL retrieved from the queue, and it prints how long that took.

The example_5.py code is listed below:

 1importqueue
 2
 3importrequests
 4fromcodetimingimport Timer
 5
 6deftask(name, work_queue):
 7    timer = Timer(text=f"Task {name} elapsed time: {{:.1f}}")
 8    with requests.Session() as session:
 9        while not work_queue.empty():
10            url = work_queue.get()
11            print(f"Task {name} getting URL: {url}")
12            timer.start()
13            session.get(url)
14            timer.stop()
15            yield
16
17defmain():
18    urls = [
19        "https://www.google.com",
20        "https://www.linkedin.com",
21        "https://www.apple.com",
22        "https://www.microsoft.com",
23        "https://www.facebook.com",
24        "https://x.com",
25    ]
26
27    work_queue = queue.Queue()
28    for url in urls:
29        work_queue.put(url)
30
31    tasks = [task("One", work_queue), task("Two", work_queue)]
32
33    with Timer(text="\nTotal elapsed time: {:.1f}"):
34        while tasks:
35            for current_task in tasks.copy():
36                try:
37                    next(current_task)
38                except StopIteration:
39                    tasks.remove(current_task)
40
41if __name__ == "__main__":
42    main()

Here’s what’s happening in this program:

• Line 3 imports requests, which provides a convenient way to make HTTP calls.
• Line 4 imports Timer from the codetiming module.
• Line 7 creates the Timer instance that measures how long each iteration of the task loop takes.
• Line 8 opens a requests session context manager, which reuses one connection pool across every URL the task fetches.
• Line 12 starts the timer instance.
• Line 13 introduces a delay, similar to example_3.py. This time it calls session.get(url), which fetches the URL retrieved from work_queue.
• Line 14 stops the timer instance and outputs the elapsed time since timer.start() was called.
• Lines 18 to 25 gather the URLs you want to fetch into a list.
• Lines 28 to 29 put those URLs into work_queue.
• Line 33 creates a Timer context manager that outputs how long the entire run took.

When you run this program, you’ll see the following output:

$ pythonexample_5.py
Task One getting URL: https://www.google.com
Task One elapsed time: 0.1
Task Two getting URL: https://www.linkedin.com
Task Two elapsed time: 0.3
Task One getting URL: https://www.apple.com
Task One elapsed time: 1.2
Task Two getting URL: https://www.microsoft.com
Task Two elapsed time: 0.3
Task One getting URL: https://www.facebook.com
Task One elapsed time: 0.5
Task Two getting URL: https://x.com
Task Two elapsed time: 0.3

Total elapsed time: 2.7

Just like in earlier versions of the program, yield turns task() into a generator. It also performs a context switch that lets the other task instance run.

Each task gets a URL from the work queue, retrieves the contents of the page, and reports how long it took to get that content.

As before, yield allows both your tasks to run cooperatively. However, since this program is running synchronously, each session.get() call blocks the CPU until the page is retrieved. At the end, note the total time it took to run the entire program. This will be meaningful for the next example.

Asynchronous (Non-Blocking) HTTP Calls

This version of the program modifies the previous one to use Python async features. It also imports aiohttp, a library for making HTTP requests asynchronously with asyncio. HTTPX is another popular choice, with an API close to the one requests gives you.

The tasks here drop the yield statement, since the code that makes the HTTP GET call no longer blocks. Awaiting the request performs the context switch back to the event loop instead.

Each request also runs inside asyncio.timeout(), so an unresponsive server can’t stall a task indefinitely. If the block takes longer than ten seconds, then asyncio.timeout() raises a TimeoutError.

The example_6.py program is listed below:

 1importasyncio
 2
 3importaiohttp
 4fromcodetimingimport Timer
 5
 6async deftask(name, work_queue):
 7    timer = Timer(text=f"Task {name} elapsed time: {{:.1f}}")
 8    async with aiohttp.ClientSession() as session:
 9        while not work_queue.empty():
10            url = await work_queue.get()
11            print(f"Task {name} getting URL: {url}")
12            timer.start()
13            async with asyncio.timeout(10):
14                async with session.get(url) as response:
15                    await response.text()
16            timer.stop()
17
18async defmain():
19    urls = [
20        "https://www.google.com",
21        "https://www.linkedin.com",
22        "https://www.apple.com",
23        "https://www.microsoft.com",
24        "https://www.facebook.com",
25        "https://x.com",
26    ]
27
28    work_queue = asyncio.Queue()
29    for url in urls:
30        await work_queue.put(url)
31
32    with Timer(text="\nTotal elapsed time: {:.1f}"):
33        async with asyncio.TaskGroup() as group:
34            group.create_task(task("One", work_queue))
35            group.create_task(task("Two", work_queue))
36
37if __name__ == "__main__":
38    asyncio.run(main())

Here’s what’s happening in this program:

• Line 3 imports the aiohttp library, which provides an asynchronous way to make HTTP calls.
• Line 4 imports Timer from the codetiming module.
• Line 6 marks task() as an async function.
• Line 7 creates the Timer instance that measures how long each request takes.
• Line 8 creates an aiohttp session context manager.
• Line 12 starts the timer instance.
• Line 13 puts a ten-second ceiling on the request that follows.
• Line 14 creates an aiohttp response context manager and makes the HTTP GET call to the URL taken from work_queue.
• Line 15 awaits the response body, handing control back to the event loop while the data arrives.
• Line 16 stops the timer instance and outputs the elapsed time since timer.start() was called.
• Lines 19 to 26 gather the same URLs you fetched synchronously a moment ago.
• Lines 29 to 30 put those URLs into work_queue, this time awaiting each put().
• Line 32 creates a Timer context manager that outputs how long the entire run took.
• Lines 33 to 35 open a task group and schedule both tasks in it.
• Line 38 starts the program running asynchronously, which also starts the internal event loop.

When you run this program, you’ll see the following output:

$ pythonexample_6.py
Task One getting URL: https://www.google.com
Task Two getting URL: https://www.linkedin.com
Task One elapsed time: 0.1
Task One getting URL: https://www.apple.com
Task Two elapsed time: 0.3
Task Two getting URL: https://www.microsoft.com
Task One elapsed time: 0.9
Task One getting URL: https://www.facebook.com
Task Two elapsed time: 0.9
Task Two getting URL: https://x.com
Task One elapsed time: 0.4
Task Two elapsed time: 0.3

Total elapsed time: 1.4

Take a look at the total elapsed time, as well as the individual times to get the contents of each URL. You’ll see that the duration is about half the cumulative time of all the HTTP GET calls. This is because the HTTP GET calls are running asynchronously. In other words, you’re effectively taking better advantage of the CPU by allowing it to make multiple requests at once.

This is where the single-threaded model stops feeling like a limitation. Your Python code still executes one line at a time, but the requests themselves don’t wait their turn. Once a task awaits a response, that request stays in flight while another task runs, so all six can genuinely be traveling the network at the same moment.

Because the CPU is so fast, this example could create as many tasks as there are URLs. In that case, the program’s runtime would be that of the single slowest URL retrieval.