Python 迭代器、生成器与上下文管理：让数据流和资源都有边界

from collections.abc import Iterable, Iterator


def nonblank_lines(lines: Iterable[str]) -> Iterator[str]:
    for line in lines:
        cleaned = line.strip()
        if cleaned:
            yield cleaned


def unique_in_order(values: Iterable[str]) -> Iterator[str]:
    seen: set[str] = set()
    for value in values:
        if value not in seen:
            seen.add(value)
            yield value

import unittest


class IteratorGuideTests(unittest.TestCase):
    def test_generator_is_lazy_and_one_shot(self):
        events: list[str] = []

        def source():
            events.append("started")
            yield "  alpha  "
            yield "   "
            yield "beta"

        values = nonblank_lines(source())
        self.assertIs(iter(values), values)
        self.assertEqual(events, [])
        self.assertEqual(next(values), "alpha")
        self.assertEqual(events, ["started"])
        self.assertEqual(list(values), ["beta"])
        self.assertEqual(list(values), [])

    def test_unique_values_keep_first_seen_order(self):
        self.assertEqual(
            list(unique_in_order(["a", "b", "a", "c", "b"])),
            ["a", "b", "c"],
        )

    def test_generator_expression_is_lazy(self):
        seen: list[int] = []
        values = (seen.append(value) or value * 2 for value in range(3))
        self.assertEqual(seen, [])
        self.assertEqual(next(values), 0)
        self.assertEqual(seen, [0])

from collections.abc import Iterator
from contextlib import contextmanager


@contextmanager
def activity_scope(events: list[str], name: str) -> Iterator[None]:
    events.append(f"start:{name}")
    try:
        yield
    finally:
        events.append(f"finish:{name}")

from contextlib import ExitStack
from collections.abc import Iterable, Iterator
from pathlib import Path
from typing import TextIO


@contextmanager
def open_text_files(paths: Iterable[Path]) -> Iterator[list[TextIO]]:
    with ExitStack() as stack:
        files = [
            stack.enter_context(path.open("r", encoding="utf-8"))
            for path in paths
        ]
        yield files

def print_nonblank_files(paths: Iterable[Path]) -> None:
    with open_text_files(paths) as files:
        for file in files:
            for line in nonblank_lines(file):
                print(line)

import tempfile
import unittest
from pathlib import Path

from resource_guide import activity_scope, nonblank_lines, open_text_files, unique_in_order


class ResourceGuideTests(unittest.TestCase):
    def test_context_manager_cleans_up_after_normal_exit(self):
        events: list[str] = []
        with activity_scope(events, "read"):
            events.append("body")
        self.assertEqual(events, ["start:read", "body", "finish:read"])

    def test_context_manager_cleans_up_and_propagates_errors(self):
        events: list[str] = []
        with self.assertRaisesRegex(RuntimeError, "broken"):
            with activity_scope(events, "write"):
                raise RuntimeError("broken")
        self.assertEqual(events, ["start:write", "finish:write"])

    def test_exit_stack_closes_all_opened_files(self):
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            first = root / "first.txt"
            second = root / "second.txt"
            first.write_text(" alpha \n", encoding="utf-8")
            second.write_text("beta\n", encoding="utf-8")

            with open_text_files([first, second]) as files:
                self.assertEqual(
                    [list(nonblank_lines(file)) for file in files],
                    [["alpha"], ["beta"]],
                )
                self.assertTrue(all(not file.closed for file in files))

            self.assertTrue(all(file.closed for file in files))


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

python -m unittest -v test_resource_guide.py