Python Generators Explained — yield, Lazy Evaluation and Real-World Patterns

# A generator function — notice it uses 'yield', not 'return'
def count_up_to(maximum):
    current = 1
    while current <= maximum:
        # Pause here, hand 'current' to the caller, remember everything
        yield current
        # Execution resumes HERE the next time next() is called
        current += 1
    # When the while loop ends, the generator raises StopIteration automatically

# Calling the function does NOT run the body — it returns a generator object
counter = count_up_to(5)
print(type(counter))   # <class 'generator'>

# next() runs the body until the next yield, then pauses again
print(next(counter))   # 1
print(next(counter))   # 2
print(next(counter))   # 3

# A for loop calls next() behind the scenes until StopIteration is raised
for number in count_up_to(3):
    print(f"Got: {number}")

# Comparing memory: a list builds everything now; a generator builds nothing yet
import sys
eager_list = list(range(1_000_000))      # all 1M integers in memory immediately
lazy_generator = range(1_000_000)        # range is also lazy — holds almost nothing

print(f"List size:      {sys.getsizeof(eager_list):,} bytes")
print(f"Generator size: {sys.getsizeof(lazy_generator)} bytes")

# Simulating a large log file with an in-memory list so you can run this locally
# In production, replace 'simulated_log_lines' with open('server.log')
simulated_log_lines = [
    "2024-01-15 INFO  User alice logged in",
    "2024-01-15 ERROR Disk quota exceeded for user bob",
    "2024-01-15 INFO  File uploaded by alice",
    "2024-01-15 ERROR Connection timeout on port 5432",
    "2024-01-15 WARNING Memory usage at 87%",
    "2024-01-15 ERROR Failed to write to /var/log/app.log",
    "2024-01-15 INFO  Scheduled backup started",
]

# STEP 1: Generator that yields one line at a time (stripped of whitespace)
def read_lines(log_source):
    for raw_line in log_source:
        yield raw_line.strip()   # hand back one clean line, then pause

# STEP 2: Generator that filters — only passes ERROR lines downstream
def filter_errors(lines):
    for line in lines:           # pulls from whatever generator feeds it
        if "ERROR" in line:
            yield line           # only yield lines that match

# STEP 3: Generator that parses each error line into a structured dict
def parse_log_entry(lines):
    for line in lines:
        parts = line.split(maxsplit=3)   # split into at most 4 parts
        if len(parts) == 4:
            date, time, level, message = parts
            yield {              # yield a dict, not a raw string
                "timestamp": f"{date} {time}",
                "level": level,
                "message": message
            }

# BUILD THE PIPELINE — nothing runs yet, we're just wiring up the chain
raw        = read_lines(simulated_log_lines)
errors     = filter_errors(raw)
parsed     = parse_log_entry(errors)

# The pipeline only executes as we consume the final generator
print("=== Error Report ===")
error_count = 0
for entry in parsed:             # THIS is the moment the whole chain starts pulling
    print(f"[{entry['timestamp']}] {entry['message']}")
    error_count += 1

print(f"\nTotal errors found: {error_count}")

import sys

# ── GENERATOR EXPRESSION vs LIST COMPREHENSION ──────────────────────────────
file_sizes_kb = [120, 45, 890, 33, 1200, 67, 450]

# List comprehension — all results computed and stored in memory RIGHT NOW
large_files_list = [size for size in file_sizes_kb if size > 100]

# Generator expression — nothing computed yet, just a recipe
large_files_gen  = (size for size in file_sizes_kb if size > 100)

print(f"List type:      {type(large_files_list)}, size: {sys.getsizeof(large_files_list)} bytes")
print(f"Generator type: {type(large_files_gen)}, size: {sys.getsizeof(large_files_gen)} bytes")

# Generator expressions work directly inside built-in functions — no extra parens needed
total_large = sum(size for size in file_sizes_kb if size > 100)  # clean and memory-efficient
print(f"Total KB in large files: {total_large}")


# ── INFINITE SEQUENCE GENERATOR ─────────────────────────────────────────────
def fibonacci_forever():
    """Yields Fibonacci numbers indefinitely — you decide when to stop."""
    previous, current = 0, 1
    while True:             # infinite loop is fine here — yield suspends, not crashes
        yield current
        previous, current = current, previous + current

# Use islice to take a finite slice from an infinite generator
from itertools import islice

first_ten_fibs = list(islice(fibonacci_forever(), 10))
print(f"\nFirst 10 Fibonacci numbers: {first_ten_fibs}")


# ── send() — TWO-WAY COMMUNICATION WITH A GENERATOR ─────────────────────────
def running_total_tracker():
    """A generator that keeps a running total of values sent into it."""
    total = 0
    while True:
        # yield sends the current total OUT, and receives the next amount IN
        amount_received = yield total
        if amount_received is not None:  # first next() sends None, so guard against it
            total += amount_received

tracker = running_total_tracker()
next(tracker)                    # must call next() first to advance to the first yield

print(f"\nRunning total after £50:   {tracker.send(50)}")
print(f"Running total after £30:   {tracker.send(30)}")
print(f"Running total after £120:  {tracker.send(120)}")
tracker.close()                  # explicitly close the generator when done

# Demonstrating the single-use nature of generators — a common gotcha
def squares_up_to(limit):
    for n in range(1, limit + 1):
        yield n ** 2

squares_gen = squares_up_to(5)

# First pass — works perfectly
print("First iteration:")
for value in squares_gen:
    print(value, end=" ")

print("\n\nSecond iteration (same generator object):")
for value in squares_gen:    # generator is exhausted — loop body NEVER runs
    print(value, end=" ")
print("(nothing printed — generator was already consumed)")

# FIX: if you need multiple passes, regenerate or use a list
squares_list = list(squares_up_to(5))  # materialise into a list once
print("\nUsing a list — second iteration works:")
for value in squares_list:
    print(value, end=" ")
print()
for value in squares_list:             # works again — lists are reusable
    print(value, end=" ")


# When a list is clearly the right choice: sorting requires all values
unsorted_temps = [23.1, 18.9, 31.4, 15.2, 28.7]
# sorted() accepts a generator but must consume it entirely to sort — same memory cost
sorted_temps = sorted(temp for temp in unsorted_temps)   # generator here buys nothing
print(f"\n\nSorted temperatures: {sorted_temps}")
# Use a list comprehension instead when you know you'll materialise anyway
sorted_temps_clear = sorted([temp for temp in unsorted_temps])  # intent is explicit