Master Async Python: Boost Performance with Coroutines and Event Loops

If you have ever stared at a spinning loading icon and silently prayed for it to finish, you are not alone; we have all experienced that helpless moment when an app feels as slow as a 1999 program.

When you first heard the term asynchronous programming , it probably sounded like some mysterious high‑tech trick, but once you master it, you will find it not scary at all and even become a powerful tool that makes your applications run faster and smoother.

Today we will talk about asynchronous Python in a simple, fun way without boring textbook language.

Why async? Speed is everything! 🏎️

Imagine ordering food with a delivery app. After placing an order, the system must simultaneously fetch the restaurant menu, calculate delivery time, and send an order notification.

Fetch the menu

Calculate delivery time

Send order notification

If these tasks are executed one by one, the user may have to wait until hunger reaches a critical level. With asynchronous programming , these operations can run concurrently , resulting in faster response times and a better user experience.

What is async Python? 🧙‍♂️

The essence of async Python is not magic but clever multitasking. Its core ideas can be summed up in two points:

Coroutines : special tasks that can be paused and resumed, similar to browser tabs you can switch between.

Event loop : a backstage conductor that schedules and switches between coroutines.

Event loop: the behind‑the‑scenes hero 🌀

In async Python, the event loop is the core system. Its workflow can be simplified as:

Schedule tasks : put coroutines into a to‑do list.

Switch tasks : when a coroutine is paused (e.g., waiting for network I/O), the loop switches to another coroutine.

Complete tasks : once a coroutine finishes, it is removed from the list until all tasks are done.

A simple example demonstrates this:

<code>import asyncio

async def order_food():
    print("🍔 正在下单……")
    await asyncio.sleep(2)
    print("✅ 食物已下单！")

async def call_driver():
    print("🚗 正在呼叫司机……")
    await asyncio.sleep(3)
    print("✅ 司机已分配！")

async def main():
    await asyncio.gather(order_food(), call_driver())

asyncio.run(main())</code>

Here order_food and call_driver are two coroutines that start together; the event loop switches between them during their await pauses, allowing both tasks to finish without wasting time.

Real‑world example: real‑time stock price tracker 📈

Suppose you want to build a tool that tracks multiple stock prices in real time. A synchronous version would look like this:

<code>import requests

def fetch_price(stock):
    response = requests.get(f"https://api.example.com/price/{stock}")
    return response.json()["price"]

stocks = ["AAPL", "TSLA", "AMZN"]
for stock in stocks:
    print(f"{stock}: ${fetch_price(stock)}")</code>

The problem is each request waits for the previous one, so a slow API drags the whole program down.

Rewriting it with async makes the code much more efficient:

<code>import aiohttp
import asyncio

async def fetch_price(stock, session):
    async with session.get(f"https://api.example.com/price/{stock}") as response:
        data = await response.json()
        return f"{stock}: ${data['price']}"

async def main():
    stocks = ["AAPL", "TSLA", "AMZN"]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_price(stock, session) for stock in stocks]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(result)

asyncio.run(main())</code>

Now all price requests are sent simultaneously , greatly improving throughput.

Tips: avoid async pitfalls 💡

Async programming is powerful but not without traps. Below are common pitfalls and how to avoid them.

1. Misusing await

Problem : Frequent await calls in each task can degrade performance because each await pauses the coroutine and triggers a context switch.

If tasks are independent, use asyncio.gather or asyncio.create_task to run them concurrently instead of awaiting each one sequentially.

Solution :

<code># Wrong: tasks run one after another
async def process_tasks():
    await task1()
    await task2()
    await task3()

# Correct: run tasks concurrently
async def process_tasks():
    await asyncio.gather(task1(), task2(), task3())</code>

2. Not closing async resources

Problem : Forgetting to close resources such as aiohttp.ClientSession can cause leaks.

Solution : Use async with to ensure proper cleanup.

<code>async def fetch_data():
    async with aiohttp.ClientSession() as session:
        async with session.get("https://example.com") as response:
            data = await response.text()
            return data</code>

3. Ignoring exception handling

Problem : Exceptions in async tasks may be swallowed, especially when using asyncio.gather or create_task .

Solution : Wrap calls in try...except or use return_exceptions=True with asyncio.gather .

<code>try:
    await asyncio.gather(task_with_error(), another_task())
except Exception as e:
    print(f"Caught exception: {e}")</code>

4. Misunderstanding asyncio.gather behavior

Problem : If one task fails, asyncio.gather raises an exception and cancels the others.

Solution : Use return_exceptions=True to let all tasks finish.

<code>results = await asyncio.gather(task1(), task2(), return_exceptions=True)
print(results)  # e.g., [1, ValueError('task failed')]</code>

5. Nested event loops

Problem : Calling asyncio.run inside an already running loop raises RuntimeError: This event loop is already running , common in notebooks.

Solution : Use await directly in interactive environments or apply nest_asyncio (not recommended for production).

<code>import nest_asyncio
nest_asyncio.apply()</code>

6. Tasks not cancelled properly

Problem : Long‑running tasks may need cancellation, but improper handling can leave resources hanging.

Solution : Catch asyncio.CancelledError or use task.cancel() correctly.

<code>async def long_running_task():
    try:
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("Task was cancelled")
        raise

async def main():
    task = asyncio.create_task(long_running_task())
    await asyncio.sleep(1)
    task.cancel()
    await task
</code>

7. Misusing thread pools

Problem : Async code is ideal for I/O‑bound work, but CPU‑bound tasks can block the event loop if run directly.

Solution : Offload CPU‑heavy work to a thread pool using asyncio.to_thread or ThreadPoolExecutor .

<code>def cpu_intensive_task():
    return sum(range(10**6))

async def main():
    result = await asyncio.to_thread(cpu_intensive_task)
    print(result)
asyncio.run(main())</code>

8. Not handling coroutine lifecycle

Problem : Creating a coroutine without awaiting it or adding it to the loop means it never runs.

Solution : Ensure every coroutine is either awaited or scheduled with asyncio.create_task .

<code># Wrong: coroutine never runs
my_task()

# Correct:
await my_task()
asyncio.create_task(my_task())
</code>

Summary ✨

Asynchronous programming may seem confusing at first, but once you take the first step, you will find it manageable and powerful, making your code more efficient and modern.

Are you ready to try async programming, or have you already built something cool with it? Share your experience!