Advanced Techniques
Coroutine Generators
Basic Coroutine Structure
def coroutine_example():
while True:
x = yield
print(f"Received: {x}")
## Coroutine usage
coro = coroutine_example()
next(coro) ## Prime the coroutine
coro.send(10)
Generator Delegation
Yield From Mechanism
def subgenerator():
yield 1
yield 2
yield 3
def delegating_generator():
yield from subgenerator()
yield from range(4, 7)
Asynchronous Generators
Async Generator Pattern
import asyncio
async def async_generator():
for i in range(3):
await asyncio.sleep(1)
yield i
async def main():
async for value in async_generator():
print(value)
Generator State Management
graph TD
A[Generator Created] --> B[Initial State]
B --> C[Yield Values]
C --> D[Suspended State]
D --> E[Resumed]
E --> F[Completed/Exhausted]
Advanced Generator Techniques
Technique |
Description |
Use Case |
Coroutines |
Two-way communication |
Complex data processing |
Generator Delegation |
Nested generators |
Composing generator workflows |
Async Generators |
Asynchronous iteration |
I/O-bound operations |
Context Management
class GeneratorContext:
def __init__(self, gen):
self.gen = gen
def __enter__(self):
return next(self.gen)
def __exit__(self, *args):
try:
next(self.gen)
except StopIteration:
pass
def context_generator():
yield 1
yield 2
Error Handling in Generators
def error_handling_generator():
try:
yield 1
yield 2
raise ValueError("Intentional error")
except ValueError:
yield "Error occurred"
Generator Pipelining
def pipeline_generator():
def stage1():
for i in range(10):
yield i * 2
def stage2(input_gen):
for value in input_gen:
yield value + 1
result = stage2(stage1())
LabEx Advanced Generator Patterns
At LabEx, we recommend exploring these advanced generator techniques to create more flexible and efficient data processing workflows.
Complex Generator Composition
def generator_composer(*generators):
for gen in generators:
yield from gen
## Usage
gen1 = (x for x in range(3))
gen2 = (x for x in range(3, 6))
composed_gen = generator_composer(gen1, gen2)
Memory-Efficient Data Processing
Large File Processing
def file_line_generator(filename):
with open(filename, 'r') as file:
for line in file:
yield line.strip()
Infinite Generators
def infinite_counter():
num = 0
while True:
yield num
num += 1