🕯️ Magic Note
Generators and itertools are a match made in heaven. The itertools module provides dozens of tools for working with iterators and generators: chain, cycle, islice, takewhile, and many more. These tools turn generators into a complete streaming data processing toolkit.
Python
def numbers(start, end):
“””Yield numbers from start to end.”””
for i in range(start, end + 1):
yield i
def letters():
“””Yield letters A to E.”””
for ch in “ABCDE”:
yield ch
def combine():
“””Combine multiple generators.”””
yield from numbers(1, 5)
yield from letters()
yield from numbers(6, 10)
for item in combine():
print(item, end=” “)
# 1 2 3 4 5 A B C D E 6 7 8 9 10
🕯️ Magic Note
yield from also handles the send() and throw() methods automatically. This makes it superior to manual for item in subgen: yield item. Always use yield from when delegating to another generator.
Python
import itertools
# chain: combine multiple iterables
result = list(itertools.chain([1, 2, 3], [4, 5, 6], [7, 8, 9]))
print(result) # [1, 2, 3, 4, 5, 6, 7, 8, 9]
# cycle: repeat an iterable infinitely
colors = itertools.cycle([“red”, “green”, “blue”])
for _ in range(7):
print(next(colors), end=” “)
# red green blue red green blue red
# islice: slice an iterator (lazy)
infinite_gen = itertools.count(1) # infinite counter
first_ten = list(itertools.islice(infinite_gen, 10))
print(first_ten) # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# takewhile: take items while condition is true
numbers = itertools.count(1)
less_than_10 = list(itertools.takewhile(lambda x: x < 10, numbers))
print(less_than_10) # [1, 2, 3, 4, 5, 6, 7, 8, 9]
# dropwhile: skip items while condition is true
numbers = [1, 2, 3, 4, 5, 1, 2, 3]
after_three = list(itertools.dropwhile(lambda x: x < 3, numbers))
print(after_three) # [3, 4, 5, 1, 2, 3]
# groupby: group consecutive identical items
data = [1, 1, 2, 2, 2, 3, 1, 1]
for key, group in itertools.groupby(data):
print(f”{key}: {list(group)}”)
# 1: [1, 1]
# 2: [2, 2, 2]
# 3: [3]
# 1: [1, 1]
Python
import itertools
import re
from datetime import datetime
def read_log_lines(filename):
“””Generator that yields lines from a log file.”””
with open(filename, “r”) as f:
for line in f:
yield line.strip()
def filter_by_level(lines, level=”ERROR”):
“””Yield only lines with specific log level.”””
pattern = re.compile(rf”{level}\b”)
for line in lines:
if pattern.search(line):
yield line
def parse_log_line(lines):
“””Parse each log line into a structured dictionary.”””
log_pattern = re.compile(r”(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (\w+) (.+)”)
for line in lines:
match = log_pattern.match(line)
if match:
yield {
“timestamp”: datetime.fromisoformat(match.group(1)),
“level”: match.group(2),
“message”: match.group(3)
}
def extract_errors_by_hour(parsed_lines):
“””Group errors by hour.”””
for line in parsed_lines:
hour = line[“timestamp”].strftime(“%Y-%m-%d %H:00”)
yield hour, line[“message”]
# Build the pipeline
lines = read_log_lines(“application.log”)
errors = filter_by_level(lines, “ERROR”)
parsed = parse_log_line(errors)
by_hour = extract_errors_by_hour(parsed)
# Process the pipeline
for hour, message in itertools.islice(by_hour, 10):
print(f”{hour}: {message}”)
🕯️ Magic Note
This pipeline never loads the entire log file into memory. It reads one line, processes it through all stages, yields the result, then moves to the next line. Memory usage stays constant regardless of file size.
Python
def running_average():
“””Coroutine that computes running average of sent values.”””
total = 0
count = 0
average = None
while True:
value = yield average # Receive value, yield average
if value is not None:
total += value
count += 1
average = total / count
avg = running_average()
next(avg) # Prime the coroutine (advance to 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
Python
from functools import wraps
def coroutine(func):
“””Decorator that primes a coroutine.”””
@wraps(func)
def primer(*args, **kwargs):
gen = func(*args, **kwargs)
next(gen) # Prime it
return gen
return primer
@coroutine
def running_average():
total = 0
count = 0
average = None
while True:
value = yield average
total += value
count += 1
average = total / count
avg = running_average() # Already primed!
print(avg.send(10)) # 10.0
print(avg.send(20)) # 15.0
🕯️ Magic Note
This pattern is common in asynchronous programming and data processing pipelines. Coroutines are the foundation of Python’s asyncio library (though asyncio uses async/await syntax for modern coroutines).
Python
@coroutine
def source(target):
“””Source: generates data and sends to target.”””
for i in range(1, 11):
target.send(i)
target.close()
@coroutine
def multiply_by(factor, target):
“””Processing stage: multiplies by factor.”””
while True:
value = yield
target.send(value * factor)
@coroutine
def subtract(subtract_value, target):
“””Processing stage: subtracts a value.”””
while True:
value = yield
target.send(value – subtract_value)
@coroutine
def sink():
“””Sink: final destination that prints results.”””
while True:
value = yield
print(f”Result: {value}”)
# Build the pipeline
end = sink()
sub = subtract(5, end)
mult = multiply_by(2, sub)
source(mult)
# Output:
# Result: -3 (1*2 – 5)
# Result: -1 (2*2 – 5)
# Result: 1 (3*2 – 5)
# … up to 10
Python
import itertools
def number_gen():
for i in range(1, 6):
print(f”Generating {i}”)
yield i
# Create two independent copies of the generator
original = number_gen()
gen1, gen2 = itertools.tee(original, 2)
# Use both copies independently
print(“First copy (squares):”)
for x in gen1:
print(f” {x}^2 = {x ** 2}”)
print(“\nSecond copy (cubes):”)
for x in gen2:
print(f” {x}^3 = {x ** 3}”)
# Note: the generator values are generated once and cached
Python
def flatten(nested):
“””Recursively flatten nested iterables.”””
for item in nested:
if isinstance(item, (list, tuple)):
yield from flatten(item)
else:
yield item
data = [1, [2, 3], [4, [5, 6], 7], 8]
flat = list(flatten(data))
print(flat) # [1, 2, 3, 4, 5, 6, 7, 8]
# With any iterable, not just lists
def flatten_any(nested):
try:
for item in nested:
try:
# Try to iterate (if it is iterable)
for sub in flatten_any(item):
yield sub
except TypeError:
# Not iterable, yield as is
yield item
except TypeError:
yield nested
Python
import itertools
def batch_generator(iterable, batch_size=100):
“””Yield batches of items from an iterable.”””
iterator = iter(iterable)
while True:
batch = list(itertools.islice(iterator, batch_size))
if not batch:
break
yield batch
# Process a large dataset in batches
def process_large_data(data_source):
for batch in batch_generator(data_source, batch_size=1000):
# Process each batch
results = [process_item(item) for item in batch]
# Perhaps write batch results to database
save_batch(results)
# Example: batch processing a large CSV
import csv
def csv_batches(filename, batch_size):
with open(filename, “r”) as f:
reader = csv.DictReader(f)
batch = []
for row in reader:
batch.append(row)
if len(batch) >= batch_size:
yield batch
batch = []
if batch:
yield batch
Python
import requests
def fetch_paginated_api(base_url, page_size=100):
“””Generator that yields items from a paginated API lazily.”””
page = 1
while True:
response = requests.get(
base_url,
params={“page”: page, “page_size”: page_size}
)
if response.status_code != 200:
break
data = response.json()
items = data.get(“items”, [])
if not items:
break
for item in items:
yield item
has_more = data.get(“has_more”, False)
if not has_more:
break
page += 1
# Use the generator
api_gen = fetch_paginated_api(“https://api.example.com/users”)
# Process first 1000 users without loading all into memory
for user in itertools.islice(api_gen, 1000):
process_user(user)
- Forgetting to prime coroutines before using send()
- Using tee on large generators without understanding memory costs
- Not handling StopIteration when using next() directly
- Creating generator pipelines that are too deep (performance overhead)
- Using yield from on non-iterables (TypeError)
- Modifying a generator while iterating over it (not supported)
- Write a generator that yields prime numbers (infinite).
- What does itertools.tee do?
- Why is yield from better than a manual loop?
- Write a coroutine that computes a running standard deviation.
- How would you process a 10GB CSV file with limited memory?
- What is the difference between send() and next()?
⚡ Whisper
Playing with generators is like learning to conduct a river. The data flows. Your generators are gates, filters, and channels. Each generator does one small transformation, then passes the water on. The river never stops. Items come one by one, flow through the pipeline, and emerge transformed. Memory stays shallow. Speed remains high. This is stream processing. This is functional composition. This is the generator way. Once you master this, you will see opportunities everywhere. Processing logs, reading large files, handling API pagination, creating infinite sequences. All become simple, elegant, and memory-efficient. The river flows. You only need to dip your cup. Play with generators. Build pipelines. Watch the data flow. This is not just coding. This is orchestration.
