迭代器、生成器与上下文管理器

class Counter:
    """一个简单的迭代器，计数到指定上限"""

    def __init__(self, limit: int):
        self.limit = limit
        self.current = 0

    def __iter__(self):
        return self  # 迭代器必须返回自身

    def __next__(self):
        if self.current >= self.limit:
            raise StopIteration  # 迭代结束
        self.current += 1
        return self.current


counter = Counter(3)
for num in counter:
    print(num)
# 输出: 1, 2, 3

from collections.abc import Iterator, Iterable

numbers = [1, 2, 3]

print(isinstance(numbers, Iterable))  # True —— 列表是可迭代的
print(isinstance(numbers, Iterator))  # False —— 但它不是迭代器

it = iter(numbers)  # 获取迭代器
print(isinstance(it, Iterator))       # True
print(next(it))      # 1
print(next(it))      # 2
print(next(it))      # 3
# print(next(it))   # StopIteration

def count_up_to(limit: int):
    """计数生成器"""
    current = 0
    while current < limit:
        current += 1
        yield current  # 暂停函数，返回值


for num in count_up_to(5):
    print(num)
# 1, 2, 3, 4, 5

# 列表：一次性生成所有数据，占用内存
def squares_list(n: int):
    return [x ** 2 for x in range(n)]

# 生成器：惰性求值，按需生成
def squares_gen(n: int):
    for x in range(n):
        yield x ** 2

import sys

list_mem = sum(sys.getsizeof(x) for x in squares_list(1000000))
# 列表占用大量内存

gen = squares_gen(1000000)
print(sys.getsizeof(gen))  # 很小的固定开销（生成器对象本身）

def flat(nested):
    """展平嵌套列表"""
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from flat(item)  # 递归展平
        else:
            yield item

nested = [1, [2, 3], [4, [5, 6]], 7]
print(list(flat(nested)))  # [1, 2, 3, 4, 5, 6, 7]

def echo():
    """回声生成器：接收并返回发送的值"""
    while True:
        received = yield  # 暂停，等待外部发送值
        yield received


gen = echo()
next(gen)              # 启动生成器（到达第一个 yield）
print(gen.send("Hello"))  # Hello
next(gen)              # 继续到下一个 yield
print(gen.send("World"))  # World

# 列表推导式（立即求值，生成列表）
squares_list = [x ** 2 for x in range(1000000)]

# 生成器表达式（惰性求值，生成生成器）
squares_gen = (x ** 2 for x in range(1000000))

print(next(squares_gen))  # 0
print(next(squares_gen))  # 1

# 正确使用 with 自动关闭文件
with open("example.txt", "w") as file:
    file.write("Hello, Python!")
# 文件在此自动关闭，无论是否发生异常

class ManagedResource:
    """资源管理类"""

    def __enter__(self):
        print("获取资源")
        self.resource = "已获取的资源"
        return self.resource  # with 的 as 子句接收此值

    def __exit__(self, exc_type, exc_val, exc_tb):
        # exc_type/val/tb：异常信息（无异常时为 None）
        print("释放资源")
        if exc_type:
            print(f"发生了异常: {exc_val}")
            return False  # 重新抛出异常
        return False  # 不压制异常


with ManagedResource() as res:
    print(f"使用: {res}")
# 获取资源
# 使用: 已获取的资源
# 释放资源

from contextlib import contextmanager

@contextmanager
def managed_resource():
    """用生成器实现上下文管理器"""
    resource = "已获取的资源"
    print("获取资源")
    try:
        yield resource  # yield 前的代码在 __enter__ 执行
    finally:
        print("释放资源")  # yield 后的代码在 __exit__ 执行


with managed_resource() as res:
    print(f"使用: {res}")
# 获取资源
# 使用: 已获取的资源
# 释放资源

from contextlib import contextmanager

@contextmanager
def transaction(db):
    """数据库事务上下文管理器"""
    print("开始事务")
    try:
        yield db
        print("提交事务")
    except Exception as e:
        print(f"回滚事务: {e}")
        raise


class FakeDB:
    pass

db = FakeDB()

try:
    with transaction(db):
        print("执行 SQL...")
        raise RuntimeError("SQL 执行失败")
except RuntimeError as e:
    print(f"异常被传播: {e}")

# 忽略指定异常
from contextlib import suppress

with suppress(FileNotFoundError):
    open("nonexistent.txt").close()

# 相当于:
try:
    open("nonexistent.txt").close()
except FileNotFoundError:
    pass


# 重定向标准输出
from contextlib import redirect_stdout
import io

buffer = io.StringIO()
with redirect_stdout(buffer):
    print("Hello")
print(buffer.getvalue())  # "Hello\n"