Summary  
This chapter introduces generator-based pipelines, demonstrating how to build modular, lazy-evaluated data processing workflows by chaining generator functions that yield and send values between stages.

General domain of usage  
Data processing pipelines


import unittest
import user_code
import ast
import re   
import unittest
import importlib

class TestTask(unittest.TestCase):
    def test_pipeline_yields_correct_squared_evens(self):
        import user_code
        importlib.reload(user_code)
        input_numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
        expected = [36, 64, 100]
        result = list(user_code.number_pipeline(input_numbers))
        _dynamic_test(
            self,
            result == expected,
            f"Pipeline yields correct squared even numbers: {expected}",
            f"Expected {expected}, got {result}",
        )

    def test_pipeline_excludes_odd_numbers(self):
        import user_code
        importlib.reload(user_code)
        input_numbers = [1, 3, 5, 7, 9]
        result = list(user_code.number_pipeline(input_numbers))
        _dynamic_test(
            self,
            result == [],
            "Pipeline correctly excludes all odd numbers.",
            f"Expected no output for all-odd input, got {result}",
        )

    def test_pipeline_only_squares_greater_than_20(self):
        import user_code
        importlib.reload(user_code)
        input_numbers = [2, 4, 6]
        # 2*2=4, 4*4=16, 6*6=36, only 36 > 20
        expected = [36]
        result = list(user_code.number_pipeline(input_numbers))
        _dynamic_test(
            self,
            result == expected,
            "Pipeline yields only squared values greater than 20.",
            f"Expected {expected}, got {result}",
        )

    def test_pipeline_accepts_any_iterable(self):
        import user_code
        importlib.reload(user_code)
        def gen():
            for i in range(1, 21):
                yield i
        expected = [36, 64, 100, 144, 196, 256, 324, 400]
        result = list(user_code.number_pipeline(gen()))
        _dynamic_test(
            self,
            result == expected,
            "Pipeline works with any iterable input.",
            f"Expected {expected}, got {result}",
        )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\s{2,}", " ", text)
    text = re.sub(r"\s*([,:?])\s*", r"\1 ", text)
    return text.strip()

def change_var(code: str, var_name: str, value: str) -> str:
    tree = ast.parse(code)
    lines = code.splitlines()
    for i, node in enumerate(tree.body):
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name) and target.id == var_name:
                    start_line = node.lineno - 1
                    line = lines[start_line]
                    indent = ' ' * (len(line) - len(line.lstrip()))
                    lines[start_line] = f"{indent}{var_name} = {value}"
                    next_line = len(lines)
                    for next_node in tree.body[i+1:]:
                        if hasattr(next_node, 'lineno'):
                            next_line = next_node.lineno - 1
                            break
                    if next_line > start_line + 1:
                        lines[start_line+1:next_line] = []
                    
                    return '\\n'.join(lines)
    return code

if __name__ == "__main__":
    unittest.main()


test_main.py

Master the art of efficient iteration in Python by learning about iterators and generators. This course covers the creation, use, and advanced techniques of generators and iterators, with hands-on tasks and conceptual quizzes to reinforce your understanding.

Explore the fundamentals of Python's iteration protocol and learn how to create and use custom iterators.

Learn how to use generator functions and expressions for efficient, lazy iteration.

Master advanced generator features such as delegation, state management, and practical applications.

Challenge: Generator-Based Pipeline

解答