It’s time to supplement some knowledge about <span>asyncio</span>… The following content is extracted from the video by <span>gemini-2.5-pro</span> corresponding to <span>Corey Schafer</span>‘s video on YTB, provided as a backup for reference, and the original video is recommended for learning.
This is the second part; the first part can be found in the Python asyncio Tutorial: In-Depth Analysis from First Principles (Part One).
Chapter 6: Handling Multiple Tasks at Once
So far, we have been creating and waiting for tasks one by one. This is fine for a few tasks, but what if you have dozens, hundreds, or even thousands of tasks? Manually managing them would be a nightmare.<span>asyncio</span> provides powerful tools to run many tasks concurrently: <span>asyncio.gather</span> and the more modern <span>asyncio.TaskGroup</span>.
<span>asyncio.gather</span> – Gathering Awaitable Objects
<span>asyncio.gather</span> is a straightforward method for concurrently running a list of awaitable objects (coroutines or tasks).
Let’s take a look at <span>example_7.py</span>. We can create tasks using a list comprehension instead of creating <span>task1</span> and <span>task2</span> on different lines.
Gathering Coroutines:
# Gathering Coroutines
coroutines = [fetch_data(i) for i in range(1, 3)]
results = await asyncio.gather(*coroutines)
print(f"Coroutine results: {results}")
Here, we created a list of coroutine objects. The <span>*</span> in front of <span>coroutines</span> is crucial—it unpacks the list and passes each coroutine as a separate argument to <span>gather</span>. Then <span>gather</span> automatically wraps these coroutines into tasks and runs them.
Gathering Tasks:You can also explicitly create tasks first and then gather them.
# Gathering Tasks
tasks = [asyncio.create_task(fetch_data(i)) for i in range(1, 3)]
results = await asyncio.gather(*tasks)
print(f"Task results: {results}")
The subtle difference lies in when the tasks are scheduled. Using <span>asyncio.create_task</span>, the tasks are scheduled immediately onto the event loop. In contrast, when gathering the original coroutines, they are scheduled only when <span>gather</span> is called.
<span>asyncio.TaskGroup</span> – Structured Concurrency
<span>TaskGroup</span> was introduced in Python 3.11, providing what is known as “structured concurrency.” It offers a clearer and safer way to manage a group of tasks using an <span>async with</span> block.
# Task Group
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(fetch_data(i)) for i in range(1, 3)]
# When the context manager exits, all tasks have been awaited.
results = [task.result() for task in tasks]
print(f"Task group results: {results}")
<span>TaskGroup</span> is remarkable because the <span>async with</span> block will not exit until all tasks created within it (<span>tg.create_task</span>) are completed. This eliminates the need to manually <span>await</span> results afterward and ensures you do not accidentally leave tasks running in the background.
<span>gather</span> vs. <span>TaskGroup</span>: Key Differences in Error Handling
The main reason to prefer <span>TaskGroup</span> over <span>gather</span> is its superior error handling mechanism.
<span>asyncio.gather</span>(default behavior): If any task in the group raises an exception,<span>gather</span>will immediately cancel the<span>await</span>and propagate the first exception it encounters. However, other tasks in the group will not be canceled and will continue running in the background. These are known as “orphan tasks,” which can lead to unpredictable behavior and resource leaks.<span>asyncio.gather(return_exceptions=True)</span>: You can instruct<span>gather</span>to wait for all tasks to complete and return any exceptions as part of the result list. This can prevent orphan tasks, but requires you to manually check each item in the results to see if it is a successful result or an exception object.<span>asyncio.TaskGroup</span>: This is the safest choice. If any task within the group fails,<span>TaskGroup</span>will automatically cancel all other tasks in the group. It will then wait for them all to finish and raise an<span>ExceptionGroup</span>containing all the exceptions that occurred. This “fast failure” and cleanup behavior is more robust and predictable.
For new code, <span>asyncio.TaskGroup</span> is almost always the better choice.
Chapter 7: Real-World Example – Asynchronous Image Downloader
Now, let’s apply everything we’ve learned to a real-world problem. We have a synchronous Python script that does two things:
- Download Images: It takes a list of URLs and uses the blocking
<span>requests</span>library to download them one by one. This is an I/O-bound operation. - Process Images: It receives the downloaded images and applies an edge detection filter to each image using the
<span>Pillow</span>library. This involves a lot of pixel operations and is a CPU-bound operation.
Running this script synchronously takes about 23.5 seconds. Let’s see how we can speed it up.
Step 1: Performance Analysis to Identify Bottlenecks
Before optimizing, we need to know where the time is being spent. Using a performance analysis tool like <span>scalene</span> (<span>uv run -m scalene real_world_example_sync_v1.py</span>), we can analyze the performance of the code.
The analysis report confirmed our assumptions:
<span>download_single_image</span>spends most of its time in “System,” meaning it is waiting for external resources (network). This is a perfect candidate for concurrency.<span>process_single_image</span>spends most of its time in “Python,” meaning the CPU is actively performing numerical computations. This is a CPU-bound task.
Step 2: Choosing the Right Tools for the Tasks
Based on our performance analysis, we can formulate a strategy:
- Downloading (I/O-bound): We can run these tasks concurrently. Since the
<span>requests</span>library is blocking, we need to run each download in a separate thread. A better long-term solution is to use a native asynchronous HTTP library like<span>httpx</span>. - Processing (CPU-bound): To achieve true parallelism and bypass Python’s Global Interpreter Lock (GIL), we need to run this work in separate processes.
Step 3: Refactoring the Code
Let’s convert the synchronous script into a high-performance asynchronous script.
1. Convert the main function to <span>async</span> First, we will turn the <span>download_images</span>, <span>process_images</span>, and <span>main</span> functions into coroutines by adding the <span>async</span> keyword. We will use <span>asyncio.run(main())</span> to run the top-level <span>main</span> function.
2. Use <span>httpx</span> and <span>aiofiles</span> for Asynchronous Downloads We will no longer run the blocking <span>requests</span> library in threads but will switch to using <span>httpx</span> and <span>aiofiles</span> for native asynchronous performance.
# pip install httpx aiofiles
import httpx
import aiofiles
async def download_single_image(client, url, ...):
response = await client.get(url, timeout=10, follow_redirects=True)
response.raise_for_status()
# ... filename logic ...
async with aiofiles.open(download_path, "wb") as f:
async for chunk in response.aiter_bytes():
await f.write(chunk)
Note that we now <span>await</span> the network request (<span>client.get</span>) and each file write (<span>f.write</span>). We even use an <span>async for</span> loop to iterate over the incoming response data.
3. Use <span>Semaphore</span> to Control Concurrency We do not want to send thousands of requests at once. We can use <span>asyncio.Semaphore</span> to limit concurrency.
DOWNLOAD_LIMIT = 5
dl_semaphore = asyncio.Semaphore(DOWNLOAD_LIMIT)
async def download_single_image(client, url, ..., semaphore):
async with semaphore: # This will wait if there are already 5 downloads in progress
# ... download logic above ...
Now, at most, only 5 download tasks will run simultaneously.
4. Use <span>ProcessPoolExecutor</span> for CPU-bound Processing For our <span>process_images</span> function, we will use a process pool to run CPU-bound work.
from concurrent.futures import ProcessPoolExecutor
async def process_images(orig_paths):
loop = asyncio.get_running_loop()
with ProcessPoolExecutor() as executor:
tasks = [
loop.run_in_executor(executor, process_single_image, path)
for path in orig_paths
]
processed_paths = await asyncio.gather(*tasks)
return processed_paths
Final Results
By applying the correct concurrency strategy to each part of the script, the total execution time was reduced from 23.5 seconds to under 5 seconds.
- Image Downloading (I/O-bound): Handled by
<span>asyncio</span>and<span>httpx</span>, running concurrently on a single thread. - Image Processing (CPU-bound): Handled by
<span>ProcessPoolExecutor</span>, running in parallel across multiple CPU cores.
Chapter 8: Summary and Best Practices
We have covered a lot, from core concepts to a complete real-world refactor. Here’s a quick checklist to refer to when dealing with concurrency in your own projects:
- Is the work I/O-bound or CPU-bound? If unsure, use performance analysis tools.
- For I/O-bound work:
- Best Choice: Use native asynchronous libraries (
<span>httpx</span>,<span>aiofiles</span>,<span>asyncpg</span>, etc.) and<span>async</span>/<span>await</span>. - Good Choice: If you must use blocking libraries, use
<span>asyncio.to_thread</span>or<span>loop.run_in_executor</span>to run them in a thread pool.
- Use
<span>loop.run_in_executor</span>and process pools for true parallelism.
<span>asyncio.TaskGroup</span> for structured, safe, and robust task management.- Remember to
<span>await</span>your coroutines and tasks. - Never call blocking code directly within coroutines.
- Use code checking tools like Ruff during development and run your code with
<span>asyncio.run(debug=True)</span><code><span> to catch common errors.</span>
Asynchronous programming in Python is an extremely powerful tool for building high-performance applications. While the initial concepts may be a bit tricky, understanding the event loop, the basics of I/O vs. CPU-bound tasks, and the tools provided by <span>asyncio</span> will enable you to write faster, more efficient code.