Python Iterators and Iterables Explained — How Loops Really Work

# A plain list is an ITERABLE — it can produce iterators on demand
playlist = ["Bohemian Rhapsody", "Hotel California", "Stairway to Heaven"]

# iter() calls playlist.__iter__() and returns a fresh ITERATOR
playlist_iterator = iter(playlist)

# next() calls playlist_iterator.__next__() each time
print(next(playlist_iterator))  # Fetches item 1 — iterator remembers position
print(next(playlist_iterator))  # Fetches item 2 — picks up exactly where it left off
print(next(playlist_iterator))  # Fetches item 3 — last item

# The iterator is now exhausted — calling next() again raises StopIteration
try:
    print(next(playlist_iterator))
except StopIteration:
    print("Iterator exhausted — no more songs!")

# The original list is UNCHANGED — start a fresh iterator anytime
fresh_iterator = iter(playlist)
print(next(fresh_iterator))  # Back to the beginning — "Bohemian Rhapsody"

# A for-loop does ALL of this invisibly on every iteration
for song in playlist:          # Python calls iter(playlist) once here
    print(f"Now playing: {song}")  # Then calls next() on each cycle

class FileChunkIterator:
    """
    Reads a text file in fixed-size chunks.
    Useful for processing large files without loading them entirely into memory.
    """

    def __init__(self, filepath, chunk_size=5):
        self.filepath = filepath
        self.chunk_size = chunk_size  # How many lines per chunk
        self._file_handle = None      # Will hold the open file — initialised in __iter__
        self._line_buffer = []        # Accumulates lines until chunk is full

    def __iter__(self):
        # Open the file fresh each time iteration starts
        # This allows the iterator to be restarted cleanly
        self._file_handle = open(self.filepath, "r")
        self._line_buffer = []
        return self  # An iterator must return itself here

    def __next__(self):
        # Try to fill a chunk from the file
        while len(self._line_buffer) < self.chunk_size:
            raw_line = self._file_handle.readline()
            if not raw_line:  # readline() returns '' at end of file
                break
            self._line_buffer.append(raw_line.strip())

        if not self._line_buffer:  # Buffer empty — file is fully consumed
            self._file_handle.close()  # Clean up the file handle
            raise StopIteration      # Signal to the for-loop: we're done

        # Pull exactly chunk_size lines (or fewer if we're near the end)
        chunk = self._line_buffer[:self.chunk_size]
        self._line_buffer = self._line_buffer[self.chunk_size:]
        return chunk


# --- Demo: create a temporary log file and iterate over it in chunks ---
import tempfile
import os

# Write 12 fake log lines to a temp file
log_lines = [f"[INFO] Event #{i} processed successfully" for i in range(1, 13)]

with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as temp_log:
    temp_log.write("\n".join(log_lines))
    temp_filepath = temp_log.name

# Iterate over the log in chunks of 5 lines
chunk_reader = FileChunkIterator(temp_filepath, chunk_size=5)

for chunk_number, log_chunk in enumerate(chunk_reader, start=1):
    print(f"--- Chunk {chunk_number} ({len(log_chunk)} lines) ---")
    for log_entry in log_chunk:
        print(f"  {log_entry}")

os.unlink(temp_filepath)  # Clean up the temp file

import time

# Simulates a paginated REST API that returns users in pages
def mock_api_fetch(page_number, page_size=3):
    """
    Pretend this is calling requests.get('https://api.example.com/users?page=N').
    Returns a list of users for that page, or empty list when pages run out.
    """
    all_users = [
        {"id": 1, "name": "Alice Nakamura"},
        {"id": 2, "name": "Ben Okafor"},
        {"id": 3, "name": "Carmen Reyes"},
        {"id": 4, "name": "David Chen"},
        {"id": 5, "name": "Elena Petrov"},
        {"id": 6, "name": "Farhan Malik"},
        {"id": 7, "name": "Grace Osei"},
    ]
    start_index = (page_number - 1) * page_size
    return all_users[start_index : start_index + page_size]


def paginated_user_stream(page_size=3):
    """
    Generator that fetches users page-by-page.
    Callers never need to know pagination exists — they just iterate.
    """
    current_page = 1
    while True:
        print(f"  [Generator] Fetching page {current_page} from API...")
        page_data = mock_api_fetch(current_page, page_size)

        if not page_data:  # Empty page means no more data
            print("  [Generator] All pages consumed — raising StopIteration")
            return  # 'return' inside a generator raises StopIteration automatically

        for user in page_data:
            yield user  # Suspend here, hand one user to the caller

        current_page += 1  # Only advance the page after all users on it are yielded


# The caller just iterates — zero pagination logic leaks out
print("Streaming all users from API:\n")
for user in paginated_user_stream(page_size=3):
    print(f"  Processing user: {user['name']} (ID: {user['id']})")

print("\nDone — all users processed.")

# ── PART 1: The Exhaustion Trap ──────────────────────────────────────────────

team_members = ["Priya", "Jordan", "Kwame"]  # This is an ITERABLE
team_iterator = iter(team_members)            # This is an ITERATOR (stateful)

print("First loop:")
for member in team_iterator:
    print(f"  Hello, {member}")

print("\nSecond loop over the SAME iterator (iterator is exhausted):")
for member in team_iterator:  # This loop body never executes — silent failure!
    print(f"  Hello again, {member}")

print("  (nothing printed — iterator was already exhausted)")

print("\nSecond loop over the ORIGINAL LIST (always works):")
for member in team_members:  # list is an iterable — fresh iterator each time
    print(f"  Hello again, {member}")


# ── PART 2: The iter(callable, sentinel) Form ─────────────────────────────────

import io

# Simulate reading binary data in 4-byte blocks from a stream
binary_stream = io.BytesIO(b"TheCodeForge.io rocks!")

print("\nReading binary stream in 4-byte blocks:")

# iter(callable, sentinel): calls binary_stream.read(4) repeatedly
# Stops automatically when read() returns b'' (empty bytes — end of stream)
for block in iter(lambda: binary_stream.read(4), b""):
    print(f"  Block: {block}")  # Each block is exactly 4 bytes (except maybe last)