Advanced Python Generators: yield, send, and Producer-Consumer Pipelines

1. Introduction to knowledge points

Generators are one of the most underrated advanced features of Python. Many people only use it as a "lazy iterator". In fact, send()/throw()/close() Let the generator own itTwo-way communicationability,yield from then provides a generatorDelegate/coroutine combinationability. Mastering these, you can write extremely elegant and memory-saving code when processing large data streams, implementing coroutines, and building pipeline patterns.

2. Quick overview of core concepts

mechanism	effect
yield x	Output x, pause execution, retain state
gen.send(val)	Send a value from outside to inside the generator, val will become the return value of the yield expression
gen.throw(exc)	Throws exception where generator pauses
gen.close()	Throw at pause GeneratorExit, terminate the generator
yield from sub_gen	Delegate to subgenerator, automatically pass send/throw/close

3. Core code & usage examples

1. Basics: Memory-saving processing of large files line by line

def read_large_file(filepath: str):
    """Read line by line, keeping only one line in memory at any time"""
    with open(filepath, encoding="utf-8") as f:
        for line in f:
            yield line.strip()

# use(100MB Documents are stress-free)
for line in read_large_file("big_log.txt"):
    if "ERROR" in line:
        print(line)

2. Advanced: Use send() Implementing two-way communication - simple accumulator

def running_average():
    """Dynamic accumulator: externally sending data while getting the current mean value"""
    total, count = 0.0, 0
    average = None
    while True:
        value = yield average          # yield Output the last average value, pause and wait send
        total += value
        count += 1
        average = total / count

avg = running_average()
next(avg)                             # Start the generator (go to the first yield)

print(avg.send(10))   # 10.0
print(avg.send(20))   # 15.0
print(avg.send(30))   # 20.0
print(avg.send(40))   # 25.0

⚠️ The first time you must next(gen) or gen.send(None) Push the generator to the first yield before sending the actual value.

3. Practical combat: producer-consumer coroutine model

def consumer():
    """Consumer: Continuously receives data and processes it"""
    items = []
    while True:
        item = yield
        if item is None:          # Sentinel value, terminate
            break
        items.append(item)
        print(f"[Consumption] {item}")
    return items

def producer(data, consumer_gen):
    """Producer: sends data to consumer"""
    next(consumer_gen)            # start up
    for item in data:
        consumer_gen.send(item)
    consumer_gen.send(None)       # Send termination signal

# run
c = consumer()
producer(["A", "B", "C", "D"], c)
# output: 
# [Consumption] A
# [Consumption] B
# [Consumption] C
# [Consumption] D

4. yield from ——Generator delegate, elegant combination

def sub_gen_a():
    yield "A1"
    yield "A2"

def sub_gen_b():
    yield "B1"
    yield "B2"

def main_gen():
    yield "START"
    yield from sub_gen_a()      # Delegation: automatic forwarding send/throw/close
    yield "MIDDLE"
    yield from sub_gen_b()
    yield "END"

print(list(main_gen()))
# ['START', 'A1', 'A2', 'MIDDLE', 'B1', 'B2', 'END']

yield from The real power of: it automatically send() / throw() / close() Forwarding to subgenerators allows external code to communicate directly with the innermost generator without manual passing.

5. Practical combat: Pipeline mode processing data flow

def lines(filepath):
    with open(filepath) as f:
        yield from f

def filter_lines(lines_gen, keyword):
    for line in lines_gen:
        if keyword in line:
            yield line

def parse_json_lines(lines_gen):
    import json
    for line in lines_gen:
        yield json.loads(line)

def extract_field(json_gen, field):
    for obj in json_gen:
        yield obj.get(field)

# Assembly pipeline (zero copy, lazy evaluation throughout the process))
pipe = extract_field(
    parse_json_lines(
        filter_lines(
            lines("data.jsonl"),
            "ERROR"
        )
    ),
    "timestamp"
)

for ts in pipe:
    print(f"wrong time: {ts}")

Each pipeline node is a generator, and data flows through it one by one. The memory footprint = the size of one record, regardless of the total file size.

6. Generator expressions vs list comprehensions - memory comparison

import sys

list_comp = [x ** 2 for x in range(10_000_000)]   # Build them all now
gen_expr  = (x ** 2 for x in range(10_000_000))    # Lazy, does not occupy memory

print(f"list comprehension: {sys.getsizeof(list_comp) / 1024 / 1024:.1f} MB")  # ~305 MB
print(f"generator:   {sys.getsizeof(gen_expr)} bytes")                   # ~112 byte

4. Precautions/Guidelines for avoiding pitfalls

1. Generators can only be traversed once ——The second traversal is empty. If reuse is needed, use list() Transfer the list or use itertools.tee().
2. must next() start up -- bring send() The generator of the first send() Must be passed None, otherwise throw TypeError.
3. Exception handling ——Exceptions thrown inside the generator will bubble up to the caller. You can use gen.throw() Test the generator's exception handling capabilities.
4. Close generator ——Can be called when no longer in use gen.close() Release resources and make them available in the generator try/finally Do the cleanup.
5. Don't mix return and yield —— Generator starting with Python 3.3+ return The value of will be appended to StopIteration.value, need to pass yield from capture.

def gen_with_return():
    yield 1
    yield 2
    return "Finish"

g = gen_with_return()
print(list(g))  # [1, 2] —— return value is ignored!

# Get it correctly return value way
def wrapper():
    result = yield from gen_with_return()
    print(f"subgenerator returns: {result}")

list(wrapper())  # output: subgenerator returns: Finish

5. Summary

Generators are much more than "memory-efficient iterators":

• send() —— Allow two-way dialogue between the outside and the generator to implement the coroutine mode
• yield from —— Elegant combination generator, automatically forward messages
• Pipeline mode ——Chain processing of large data streams, each record flows through the pipeline step by step

Next time you face a large amount of data processing or a scenario that requires a producer-consumer model, use a generator instead of writing a list first and then process it. Not only will the code be more elegant, but the memory performance will also be significantly improved.