Python Generators Practice Problems & Exercises
Practice: Generators and yield — Suspended Execution at Engineering Depth
← Back to lessonEasy
Write a generator count_up(start, stop) that yields integers from start to stop inclusive.
def count_up(start, stop):
n = start
while n <= stop:
yield n
n += 1
for n in count_up(3, 7):
print(n)Solution
def count_up(start, stop):
n = start
while n <= stop:
yield n
n += 1
for n in count_up(3, 7):
print(n)
# 3 4 5 6 7
Explanation: Any function containing yield becomes a generator function. Calling it returns a generator object — no code runs until the first next() call (or the first loop iteration). Each yield suspends execution and hands the value to the caller; the frame resumes at the next line when the caller requests the next value.
def count_up(start, stop):
# Yield integers from start to stop (inclusive)
pass
for n in count_up(3, 7):
print(n)Expected Output
3\n4\n5\n6\n7Hints
Hint 1: Use a while loop and yield each value before incrementing.
Hint 2: The function becomes a generator function the moment it contains a yield statement.
Write an infinite generator infinite_counter(start, step) that yields values forever. Use itertools.islice to take only the first 5.
import itertools
def infinite_counter(start=0, step=1):
n = start
while True:
yield n
n += step
result = list(itertools.islice(infinite_counter(10, 3), 5))
print(result)Solution
import itertools
def infinite_counter(start=0, step=1):
n = start
while True:
yield n
n += step
result = list(itertools.islice(infinite_counter(10, 3), 5))
print(result) # [10, 13, 16, 19, 22]
Explanation: An infinite while True loop is perfectly safe in a generator because execution suspends at each yield. The generator only advances when the caller calls next(). itertools.islice closes the generator after consuming n items, triggering GeneratorExit inside the generator.
def infinite_counter(start=0, step=1):
# Yield start, start+step, start+2*step, ... forever
pass
import itertools
# Take only the first 5 values starting at 10, step 3
result = list(itertools.islice(infinite_counter(10, 3), 5))
print(result)Expected Output
[10, 13, 16, 19, 22]Hints
Hint 1: Use while True: to loop forever — safe because generators are lazy.
Hint 2: yield the current value, then increment by step.
Run this code to observe the memory difference between a list comprehension and a generator expression over 1,000,000 elements. Note that exact byte counts may vary by Python version.
import sys
list_comp = [x * x for x in range(1_000_000)]
gen_expr = (x * x for x in range(1_000_000))
print("List size (bytes):", sys.getsizeof(list_comp))
print("Generator size (bytes):", sys.getsizeof(gen_expr))
print("First 5 from generator:", [next(gen_expr) for _ in range(5)])Solution
import sys
list_comp = [x * x for x in range(1_000_000)]
gen_expr = (x * x for x in range(1_000_000))
print("List size (bytes):", sys.getsizeof(list_comp)) # ~8 MB
print("Generator size (bytes):", sys.getsizeof(gen_expr)) # ~104 bytes
print("First 5 from generator:", [next(gen_expr) for _ in range(5)])
Explanation: The list allocates memory for all 1,000,000 integers immediately. The generator object is ~104 bytes regardless of how many values it will eventually produce — it stores only the frame pointer, the current execution state, and the local variables. This constant-memory property makes generators essential for large-scale data processing.
import sys
# Compare memory usage of a list comprehension vs a generator expression
# for the first 1,000,000 squares
list_comp = [x * x for x in range(1_000_000)]
gen_expr = (x * x for x in range(1_000_000))
print("List size (bytes):", sys.getsizeof(list_comp))
print("Generator size (bytes):", sys.getsizeof(gen_expr))
print("First 5 from generator:", [next(gen_expr) for _ in range(5)])Expected Output
List size (bytes): 8000056\nGenerator size (bytes): 104\nFirst 5 from generator: [0, 1, 4, 9, 16]Hints
Hint 1: This problem is observational — run it and note the size difference.
Hint 2: Generator expressions use () instead of [].
Hint 3: A generator object is a fixed small size regardless of how many values it will yield.
Write an infinite generator fibonacci() that yields the Fibonacci sequence starting from 0.
import itertools
def fibonacci():
a, b = 0, 1
while True:
yield a
a, b = b, a + b
print(list(itertools.islice(fibonacci(), 10)))Solution
import itertools
def fibonacci():
a, b = 0, 1
while True:
yield a
a, b = b, a + b
print(list(itertools.islice(fibonacci(), 10)))
# [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
Explanation: The generator maintains state between yields via its local variables a and b. The tuple swap a, b = b, a + b is atomic (the right side is evaluated first), so no temporary variable is needed. This pattern — an infinite stateful sequence — is the canonical use case for generators.
def fibonacci():
# Yield the Fibonacci sequence: 0, 1, 1, 2, 3, 5, 8, 13, ...
pass
import itertools
print(list(itertools.islice(fibonacci(), 10)))Expected Output
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34]Hints
Hint 1: Track two variables: a and b.
Hint 2: Each step: yield a, then update a, b = b, a+b.
Medium
Implement three generator functions and chain them into a pipeline. Each generator should yield from its predecessor — no intermediate lists.
def read_numbers(lst):
yield from lst
def filter_positive(numbers):
for n in numbers:
if n > 0:
yield n
def square_values(numbers):
for n in numbers:
yield n * n
data = [-3, -1, 0, 2, 5, -2, 8]
pipeline = square_values(filter_positive(read_numbers(data)))
print(list(pipeline))Solution
def read_numbers(lst):
yield from lst
def filter_positive(numbers):
for n in numbers:
if n > 0:
yield n
def square_values(numbers):
for n in numbers:
yield n * n
data = [-3, -1, 0, 2, 5, -2, 8]
pipeline = square_values(filter_positive(read_numbers(data)))
print(list(pipeline)) # [4, 25, 64]
Explanation: Each generator wraps the previous one. When list() calls next() on square_values, it calls next() on filter_positive, which calls next() on read_numbers. Data flows upward one element at a time — the entire chain executes in a single pass with O(1) memory regardless of input size.
def read_numbers(lst):
"""Yield each number from the list."""
pass
def filter_positive(numbers):
"""Yield only positive numbers from the input generator."""
pass
def square_values(numbers):
"""Yield the square of each number from the input generator."""
pass
data = [-3, -1, 0, 2, 5, -2, 8]
pipeline = square_values(filter_positive(read_numbers(data)))
print(list(pipeline))Expected Output
[4, 25, 64]Hints
Hint 1: Each function is a generator — use yield inside each one.
Hint 2: Chain them: square_values wraps filter_positive wraps read_numbers.
Hint 3: Each generator pulls from the one below it, element by element.
Implement a recursive flatten generator that handles arbitrarily nested lists using yield from.
def flatten(nested):
for item in nested:
if isinstance(item, list):
yield from flatten(item)
else:
yield item
data = [1, [2, 3], [4, [5, 6]], 7, [[8, 9], 10]]
print(list(flatten(data)))Solution
def flatten(nested):
for item in nested:
if isinstance(item, list):
yield from flatten(item)
else:
yield item
data = [1, [2, 3], [4, [5, 6]], 7, [[8, 9], 10]]
print(list(flatten(data))) # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
Explanation: yield from flatten(item) delegates to the recursive generator and transparently passes every yielded value up the call chain. Without yield from, you would need for x in flatten(item): yield x — more verbose and slower because each yield goes through an extra generator frame. yield from also forwards send(), throw(), and close() correctly, which matters for coroutine use cases.
def flatten(nested):
# Use yield from to recursively flatten an arbitrarily nested list.
pass
data = [1, [2, 3], [4, [5, 6]], 7, [[8, 9], 10]]
print(list(flatten(data)))Expected Output
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]Hints
Hint 1: Check if each item is a list (or more generally, an Iterable that is not a string).
Hint 2: If it is a list, yield from flatten(item) — this delegates recursively.
Hint 3: If it is a plain value, yield it directly.
Implement a running_average() coroutine. After priming with next(), each gen.send(value) should return the updated running average.
def running_average():
total = 0
count = 0
while True:
value = yield total / count if count else 0.0
total += value
count += 1
gen = running_average()
next(gen) # prime: advance to first yield
print(gen.send(10)) # 10.0
print(gen.send(20)) # 15.0
print(gen.send(30)) # 20.0
print(gen.send(40)) # 25.0Solution
def running_average():
total = 0
count = 0
while True:
value = yield total / count if count else 0.0
total += value
count += 1
gen = running_average()
next(gen)
print(gen.send(10)) # 10.0
print(gen.send(20)) # 15.0
print(gen.send(30)) # 20.0
print(gen.send(40)) # 25.0
Explanation: value = yield expr is the two-way protocol. The expression expr is the value yielded out (returned by send() to the caller). The variable value receives whatever the caller passes into the next send() call. next(gen) is equivalent to gen.send(None) and advances the generator to the first yield — this "priming" step is required before any real send().
def running_average():
# A coroutine that maintains a running average.
# Each send(value) should update the average and yield the new average.
# Prime it with next() before sending values.
total = 0
count = 0
while True:
value = yield # receives the sent value, yields the average
# your code here
gen = running_average()
next(gen) # prime the coroutine
print(gen.send(10)) # 10.0
print(gen.send(20)) # 15.0
print(gen.send(30)) # 20.0
print(gen.send(40)) # 25.0Expected Output
10.0\n15.0\n20.0\n25.0Hints
Hint 1: value = yield suspends and waits for a .send(x) call; x is assigned to value.
Hint 2: After receiving the value, update total and count, then yield the average.
Hint 3: The yield expression both sends out a value and receives the next sent value.
Hint 4: yield avg_so_far will make the next send() receive this value as the result of yield.
Implement sliding_window(iterable, n) as a generator using collections.deque with maxlen.
from collections import deque
def sliding_window(iterable, n):
it = iter(iterable)
window = deque(maxlen=n)
# fill initial window
for _ in range(n):
try:
window.append(next(it))
except StopIteration:
return
yield tuple(window)
for item in it:
window.append(item)
yield tuple(window)
print(list(sliding_window([1, 2, 3, 4, 5], 3)))Solution
from collections import deque
def sliding_window(iterable, n):
it = iter(iterable)
window = deque(maxlen=n)
for _ in range(n):
try:
window.append(next(it))
except StopIteration:
return
yield tuple(window)
for item in it:
window.append(item)
yield tuple(window)
print(list(sliding_window([1, 2, 3, 4, 5], 3)))
# [(1, 2, 3), (2, 3, 4), (3, 4, 5)]
Explanation: A deque with maxlen=n automatically discards the oldest element when a new one is appended — O(1) append and eviction. Pre-filling the first n elements handles the "not enough data" edge case. Each subsequent element shifts the window by one. itertools.pairwise in Python 3.10+ handles n=2; for arbitrary n, more-itertools.sliding_window implements the same pattern.
from collections import deque
def sliding_window(iterable, n):
# Yield tuples of n consecutive elements.
# sliding_window([1,2,3,4,5], 3) -> (1,2,3), (2,3,4), (3,4,5)
pass
print(list(sliding_window([1, 2, 3, 4, 5], 3)))Expected Output
[(1, 2, 3), (2, 3, 4), (3, 4, 5)]Hints
Hint 1: Use a deque with maxlen=n to maintain the current window.
Hint 2: Pre-fill the deque with the first n elements, then yield a tuple of it.
Hint 3: For each subsequent element: append to deque (auto-evicts oldest), yield tuple.
Hard
Implement traffic_light() as an infinite generator that cycles through GREEN, YELLOW, RED. The generator's execution position encodes the state — no state variable required.
def traffic_light():
while True:
yield 'GREEN'
yield 'YELLOW'
yield 'RED'
light = traffic_light()
for _ in range(6):
print(next(light))Solution
def traffic_light():
while True:
yield 'GREEN'
yield 'YELLOW'
yield 'RED'
light = traffic_light()
for _ in range(6):
print(next(light))
# GREEN YELLOW RED GREEN YELLOW RED
Explanation: This demonstrates the key insight about generators: the execution position is the state. There is no state variable, no enum, no match statement. The generator suspends at each yield, and when resumed it continues from exactly that point. For complex protocols with many states, this makes the code far more readable than a hand-rolled state machine with an explicit state enum.
def traffic_light():
# A generator coroutine modelling a traffic light.
# States cycle: GREEN -> YELLOW -> RED -> GREEN -> ...
# Each next() or send() call advances to the next state and yields it.
pass
light = traffic_light()
for _ in range(6):
print(next(light))Expected Output
GREEN\nYELLOW\nRED\nGREEN\nYELLOW\nREDHints
Hint 1: Use an infinite while True loop with three yield statements.
Hint 2: The generator suspends at each yield, remembering where it is in the cycle.
Hint 3: No explicit state variable is needed — the position in the code IS the state.
Implement batch(iterable, size) as a generator that yields chunks of size elements. Use itertools.islice for slicing; handle the last partial batch correctly.
import itertools
def batch(iterable, size):
it = iter(iterable)
while True:
chunk = list(itertools.islice(it, size))
if not chunk:
return
yield chunk
data = range(1, 12)
for chunk in batch(data, 4):
print(chunk)Solution
import itertools
def batch(iterable, size):
it = iter(iterable)
while True:
chunk = list(itertools.islice(it, size))
if not chunk:
return
yield chunk
data = range(1, 12)
for chunk in batch(data, 4):
print(chunk)
# [1, 2, 3, 4]
# [5, 6, 7, 8]
# [9, 10, 11]
Explanation: Calling iter() once ensures a single iterator is advanced across all islice calls. Each islice(it, size) advances it by at most size positions without resetting it. The list() materialises the chunk (necessary to check if it is empty). In Python 3.12+, itertools.batched(iterable, n) implements this pattern natively.
import itertools
def batch(iterable, size):
# Yield successive non-overlapping chunks of exactly 'size' elements.
# The last batch may be smaller if the iterable length is not divisible by size.
pass
data = range(1, 12) # 1..11
for chunk in batch(data, 4):
print(list(chunk))Expected Output
[1, 2, 3, 4]\n[5, 6, 7, 8]\n[9, 10, 11]Hints
Hint 1: Convert the iterable to an iterator once with iter().
Hint 2: itertools.islice(it, size) lazily takes at most size elements from it.
Hint 3: When islice returns an empty iterator, you are done.
Hint 4: You need to peek or materialise each slice to know if it is empty.
Implement an aggregator() coroutine that tracks running statistics, handles throw(DataError) gracefully (skip bad value, yield current stats), and prints a message on close().
class DataError(Exception):
pass
def aggregator():
count = 0
total = 0
try:
while True:
try:
value = yield (count, total, total / count if count else 0.0)
count += 1
total += value
except DataError:
yield (count, total, total / count if count else 0.0)
finally:
print("aggregator closed")
agg = aggregator()
next(agg) # prime
print(agg.send(10))
print(agg.send(20))
agg.throw(DataError, "bad value")
print(agg.send(30))
agg.close()Solution
class DataError(Exception):
pass
def aggregator():
count = 0
total = 0
try:
while True:
try:
value = yield (count, total, total / count if count else 0.0)
count += 1
total += value
except DataError:
yield (count, total, total / count if count else 0.0)
finally:
print("aggregator closed")
agg = aggregator()
next(agg)
print(agg.send(10)) # (1, 10, 10.0)
print(agg.send(20)) # (2, 30, 15.0)
agg.throw(DataError, "bad value") # (2, 30, 15.0)
print(agg.send(30)) # (3, 60, 20.0)
agg.close() # aggregator closed
Explanation: throw(exc) injects an exception at the point where the generator is currently suspended (at the yield). The inner except DataError catches it and yields the unchanged stats. close() injects GeneratorExit (a BaseException subclass), which is caught by the outer finally — this is the correct cleanup hook for generator coroutines. This full protocol (send, throw, close) is precisely what async/await builds on under the hood.
class DataError(Exception):
pass
def aggregator():
# Coroutine that:
# - Accepts numbers via send(n)
# - Yields (count, total, average) after each send
# - On throw(DataError), skips the bad value and yields current stats unchanged
# - On close(), cleans up (print "aggregator closed")
pass
agg = aggregator()
next(agg) # prime
print(agg.send(10)) # (1, 10, 10.0)
print(agg.send(20)) # (2, 30, 15.0)
agg.throw(DataError, "bad value") # should print stats without updating
print(agg.send(30)) # (3, 60, 20.0)
agg.close() # aggregator closedExpected Output
(1, 10, 10.0)\n(2, 30, 15.0)\n(2, 30, 15.0)\n(3, 60, 20.0)\naggregator closedHints
Hint 1: Use a try/except DataError block around the yield to catch throw() calls.
Hint 2: When DataError is caught, yield the current stats without updating totals.
Hint 3: Use try/finally to handle GeneratorExit (triggered by close()).
Hint 4: Structure: outer try/finally for GeneratorExit, inner try/except for DataError.