Master Python Async: From Generators and Yield to Async/Await and Real‑World Applications

1. Generator Basics: The Magic of yield

What is a generator? A generator is a special iterator that uses the yield statement to lazily produce values instead of computing them all at once.

def simple_generator():
    """Simple generator example"""
    print("Start execution")
    yield 1
    print("Continue execution")
    yield 2
    print("End execution")
    yield 3

# Using the generator
gen = simple_generator()
print("First call:")
print(next(gen))  # 输出: 开始执行 → 1
print("Second call:")
print(next(gen))  # 输出: 继续执行 → 2
print("Third call:")
print(next(gen))  # 输出: 结束执行 → 3

Memory efficiency comparison

import sys

def get_numbers_list(n):
    numbers = []
    for i in range(n):
        numbers.append(i)
    return numbers

def get_numbers_gen(n):
    for i in range(n):
        yield i

n = 1000000
list_memory = sys.getsizeof(get_numbers_list(n))
gen_memory = sys.getsizeof(get_numbers_gen(n))
print(f"List memory usage: {list_memory / 1024 / 1024:.2f} MB")
print(f"Generator memory usage: {gen_memory / 1024 / 1024:.2f} MB")

Result: List memory ~8 MB, generator memory ~0 MB.

Generator expressions

# List comprehension (eager)
squares_list = [x * x for x in range(1000000)]
# Generator expression (lazy)
squares_gen = (x * x for x in range(1000000))
print(f"List size: {sys.getsizeof(squares_list)} bytes")
print(f"Generator size: {sys.getsizeof(squares_gen)} bytes")

2. Generator Advanced: Two‑Way Communication with yield

Python 2.5 introduced yield as an expression, allowing generators to receive external data.

def interactive_generator():
    """Generator that supports two‑way communication"""
    print("Generator started")
    received = yield "Please send data"
    print(f"Received: {received}")
    received = yield "Send more data"
    print(f"Received: {received}")
    yield "Done"

gen = interactive_generator()
print(next(gen))               # 启动生成器，输出: "请发送数据"
print(gen.send("hello"))      # 输出: "收到: hello" → "请发送更多数据"
print(gen.send("world"))      # 输出: "收到: world" → "完成"

Generators can also act as simple coroutines using send(), throw(), and close():

def coroutine_example():
    """Simple coroutine example"""
    try:
        while True:
            value = yield
            print(f"Processing value: {value}")
    except GeneratorExit:
        print("Coroutine closed")
    except Exception as e:
        print(f"Exception occurred: {e}")

coro = coroutine_example()
next(coro)               # 启动协程
coro.send(10)            # 输出: 处理值: 10
coro.send(20)            # 输出: 处理值: 20
coro.throw(ValueError("test exception"))  # 输出: 发生异常: test exception
coro.close()             # 输出: 协程关闭

3. yield from : Generator Delegation

Python 3.3 added yield from to simplify nested generators.

def sub_generator():
    yield from [1, 2, 3]
    yield from (4, 5, 6)
    yield from range(7, 11)

def main_generator():
    yield from sub_generator()
    yield "Done"

for value in main_generator():
    print(value, end=" ")
# Output: 1 2 3 4 5 6 7 8 9 10 Done

4. asyncio and Native Coroutines

Python 3.4 introduced the asyncio module; Python 3.5 added the async / await syntax, providing true native coroutines.

4.1 Basic Concepts of Asynchronous Programming

import asyncio, time

async def say_after(delay, message):
    await asyncio.sleep(delay)
    print(message)
    return f"Done: {message}"

async def main():
    print(f"Start time: {time.strftime('%X')}")
    result1 = await say_after(2, "Hello")
    result2 = await say_after(1, "World")
    print(f"End time: {time.strftime('%X')}")
    print(f"Results: {result1}, {result2}")

asyncio.run(main())

4.2 Concurrent Execution of Tasks

async def concurrent_main():
    print(f"Start time: {time.strftime('%X')}")
    task1 = asyncio.create_task(say_after(2, "Hello"))
    task2 = asyncio.create_task(say_after(1, "World"))
    results = await asyncio.gather(task1, task2)
    print(f"End time: {time.strftime('%X')}")
    print(f"All results: {results}")

asyncio.run(concurrent_main())

5. Detailed async/await Syntax

5.1 Declaring Async Functions

# Traditional generator‑based coroutine
def old_coroutine():
    value = yield
    return value

# Modern async function
async def modern_coroutine():
    result = await some_async_operation()
    return result

5.2 await Expressions

async def example():
    result1 = await async_function()
    result2 = await asyncio.Future()
    task = asyncio.create_task(async_function())
    result3 = await task
    return result1, result2, result3

6. Real‑World Cases

6.1 Asynchronous HTTP Requests

import aiohttp, asyncio

async def fetch_url(session, url):
    try:
        async with session.get(url, timeout=10) as response:
            content = await response.text()
            return f"{url}: {len(content)} bytes"
    except Exception as e:
        return f"{url}: error - {e}"

async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/3",
        "https://invalid-url-test.com",
    ]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, u) for u in urls]
        results = await asyncio.gather(*tasks)
        for r in results:
            print(r)

asyncio.run(main())

6.2 Asynchronous File Operations

import aiofiles, asyncio

async def async_file_operations():
    async with aiofiles.open('test.txt', 'w') as f:
        await f.write('Hello, async world!
')
        await f.write('This is async file IO.
')
    async with aiofiles.open('test.txt', 'r') as f:
        content = await f.read()
        print("File content:")
        print(content)
    async with aiofiles.open('test.txt', 'r') as f:
        async for line in f:
            print(f"Line: {line.strip()}")

asyncio.run(async_file_operations())

6.3 Producer‑Consumer Pattern

import asyncio, random

async def producer(queue, pid):
    for i in range(5):
        item = f"product-{pid}-{i}"
        await asyncio.sleep(random.uniform(0.1, 0.5))
        await queue.put(item)
        print(f"Producer {pid} produced: {item}")
    await queue.put(None)  # termination signal

async def consumer(queue, cid):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)  # pass the signal to other consumers
            break
        print(f"Consumer {cid} consumed: {item}")
        await asyncio.sleep(random.uniform(0.2, 0.8))

async def main():
    queue = asyncio.Queue(maxsize=10)
    producers = [producer(queue, i) for i in range(3)]
    consumers = [consumer(queue, i) for i in range(2)]
    await asyncio.gather(*producers, *consumers)

asyncio.run(main())

7. Performance Comparison: Synchronous vs Asynchronous

7.1 I/O‑Intensive Tasks

import asyncio, time, requests, aiohttp

def sync_fetch(urls):
    results = []
    for url in urls:
        r = requests.get(url)
        results.append(f"{url}: {len(r.text)} bytes")
    return results

async def async_fetch(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, u) for u in urls]
        return await asyncio.gather(*tasks)

async def fetch_url(session, url):
    async with session.get(url) as resp:
        content = await resp.text()
        return f"{url}: {len(content)} bytes"

urls = ["https://httpbin.org/delay/1"] * 10
print("Synchronous version:")
start = time.time()
sync_fetch(urls)
print(f"Sync time: {time.time() - start:.2f} seconds")
print("
Asynchronous version:")
start = time.time()
asyncio.run(async_fetch(urls))
print(f"Async time: {time.time() - start:.2f} seconds")
print(f"Speedup: { (time.time() - start) / (time.time() - start):.1f}x")

8. Error Handling and Debugging

8.1 Asynchronous Exception Handling

import asyncio, random

async def risky_operation():
    await asyncio.sleep(0.1)
    if random.random() < 0.3:
        raise ValueError("Random failure")
    return "Success"

async def robust_main():
    tasks = [asyncio.create_task(risky_operation()) for _ in range(10)]
    results = []
    for task in tasks:
        try:
            res = await task
            results.append(res)
        except ValueError as e:
            print(f"Task failed: {e}")
            results.append("Failed")
    print(f"Final results: {results}")

asyncio.run(robust_main())

8.2 Debugging Techniques

import logging, asyncio
logging.basicConfig(level=logging.DEBUG)

async def debug_coroutine():
    logging.debug("Coroutine start")
    await asyncio.sleep(0.1)
    logging.debug("First step done")
    await asyncio.sleep(0.1)
    logging.debug("Second step done")
    return "Finished"

asyncio.run(debug_coroutine(), debug=True)

9. Best Practices and Common Pitfalls

9.1 Best Practices

Use async/await instead of old‑style generator‑based coroutines.

Control concurrency limits appropriately.

Handle exceptions correctly inside async code.

Set sensible timeouts for I/O operations.

Avoid blocking calls (e.g., time.sleep ) inside async functions; use await asyncio.sleep instead.

9.2 Common Pitfalls

# Pitfall: blocking the event loop
async def bad_example():
    time.sleep(1)  # ❌ blocks the loop – should use await asyncio.sleep(1)

# Pitfall: forgetting to await
async def another_bad_example():
    result = some_async_function()  # ❌ missing await
    # Correct: result = await some_async_function()

# Correct usage
async def good_example():
    await asyncio.sleep(1)  # ✅ non‑blocking
    result = await some_async_function()

10. Summary: Evolution Roadmap

Version

Feature

Significance

Python 2.2

Generators ( yield)

Introduced lazy evaluation

Python 2.5 yield expression

Enabled two‑way communication

Python 3.3 yield from Generator delegation

Python 3.4 asyncio module

Asynchronous programming framework

Python 3.5 async / await Native coroutine support

Python 3.7 asyncio.run() Simplified async entry point

Recommendation:

✅ I/O‑bound tasks : use async/await ✅ CPU‑bound tasks : combine multiprocessing with asyncio ✅ Simple lazy computations : use generators

✅ Complex async logic : leverage the asyncio ecosystem