Python 速成 Day 2：从函数包装到装饰器，从迭代到生成器

今天要达到的能力

完成 Day 1 后，你已经能判断数据是否共享。今天继续理解两件事：函数如何被包装，数据何时真正计算。

学完后应能读懂普通装饰器、闭包与生成器，预测它们的执行顺序，并写出计时包装器和分批数据处理函数。示例基于 Python 3.12，不依赖具体项目或外部服务。

时间	内容
35 分钟	函数、闭包、包装与装饰器
25 分钟	迭代协议、生成器与输出预测
50 分钟	三个编码练习
30 分钟	对照答案和面试自测
10 分钟	总结执行时机与易错点

必须掌握：函数对象、闭包、普通装饰器与 wraps、iter/next、yield 和惰性消费。知道即可：lambda、__iter__、__next__、yield from。今天跳过复杂装饰器工厂、生成器 send/throw、自定义复杂容器及异步装饰器。

1. 函数本身也是对象

Python 函数可以赋值给名称、传给另一个函数、作为返回值。这和 JS 一等函数相近，适合工具注册、回调和处理策略。

def normalize(text):
    return text.strip().lower()

def process(text, transform):
    return transform(text)

assert process("  PYTHON  ", normalize) == "python"

normalize 表示函数对象，normalize("TEXT") 表示调用函数后得到的结果。传入函数可以让 process 复用不同处理逻辑，不必每加一个策略就修改内部条件分支。

JS 中同样可以写 process(text, normalize)。Python 的 lambda 对应简短匿名函数，但函数体只能是一个表达式：

records = [{"id": "a", "score": 0.8}, {"id": "b", "score": 0.9}]
ordered = sorted(records, key=lambda record: record["score"], reverse=True)
assert ordered[0]["id"] == "b"

复杂逻辑直接用 def，比塞进 lambda 更容易阅读。

2. 返回函数后，外层变量如何继续使用

既然可以返回函数，就可以把配置与行为组合起来。内部函数访问外层作用域的名称，这就是闭包。

def make_prefixer(prefix):
    def apply(text):
        return prefix + text
    return apply

add_title = make_prefixer("标题：")
assert add_title("Python") == "标题：Python"

外层调用结束后，返回的函数仍能访问 prefix。机制上，函数保留访问自由变量绑定所需的信息，而不是必须让外层调用一直运行。

需要注意：闭包不是创建时所有值的自动快照。

def make_callbacks():
    callbacks = []
    for i in range(3):
        callbacks.append(lambda: i)
    return callbacks

assert [fn() for fn in make_callbacks()] == [2, 2, 2]

调用时读取同一个 i，此时值已是 2。JS 的 for (let i = ...) 有每轮独立绑定，不能直接套用。利用 Day 1 的默认参数求值时机，可以保存每次的整数：

def make_callbacks():
    callbacks = []
    for i in range(3):
        callbacks.append(lambda value=i: value)
    return callbacks

assert [fn() for fn in make_callbacks()] == [0, 1, 2]

这仍不深拷贝可变对象，只是保存当时绑定的对象。

3. 从高阶函数走到装饰器

现在把“接收函数”和“返回闭包”结合起来，为已有函数增加日志行为：

def log_call(func):
    def wrapper(*args, **kwargs):
        print("开始调用")
        result = func(*args, **kwargs)
        print("调用成功")
        return result
    return wrapper

def greet(name):
    return f"Hello, {name}"

greet = log_call(greet)
assert greet("Alice") == "Hello, Alice"

wrapper 通过闭包保留原函数 func，接收参数后转交给原函数，再返回结果。原函数抛异常时，“调用成功”不会执行，异常会继续传播。

Python 用装饰器语法表达这种应用关系：

def decorate(func):
    print("包装函数")
    def wrapper():
        print("调用包装器")
        return func()
    return wrapper

@decorate
def work():
    print("执行业务")

print("准备调用")
work()

输出依次是“包装函数、准备调用、调用包装器、执行业务”。这里 @decorate 可以理解为定义 work 后执行 work = decorate(work)。

执行函数定义时应用装饰器，调用新函数时执行包装器。这解释了为什么顶层函数的装饰器通常在首次导入执行模块时就生效。它适合集中处理日志、计时或注册行为；不是每个装饰器都包装调用，也可能只是登记原函数后返回它。

JS 的直接对照是 const wrapped = decorate(original)，不用先套用 TS 装饰器的具体规范。

4. 保留被包装函数的信息

包装器可以保留业务行为，却让函数名和文档变成 wrapper 的信息。调试和框架检查通常仍需要原函数的信息，因此使用 functools.wraps：

from functools import wraps

def log_call(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@log_call
def greet(name):
    """生成问候语。"""
    return f"Hello, {name}"

assert greet.__name__ == "greet"
assert greet.__doc__ == "生成问候语。"
assert greet.__wrapped__("Alice") == "Hello, Alice"

wraps 帮助保留名称、文档等元信息，并通过 __wrapped__ 指向原函数。它不会自动让同步包装器正确等待异步函数，也不会替你处理重试或异常。

到这里，装饰器没有神秘之处：函数对象让行为可传递，闭包保存原函数，包装器加入逻辑，@ 简化应用，wraps 保留元信息。

5. 让不同数据源共享同一套遍历方式

接下来从“怎样调用函数”转向“怎样逐项取得数据”。列表、文件和生成器内部结构不同，却都能用于 for，因为它们遵循迭代协议。

概念	含义
iterable，可迭代对象	能通过 iter() 得到迭代器，例如列表、字符串
iterator，迭代器	保存遍历状态，通过 next() 请求下一项
generator，生成器	一种迭代器，用暂停和恢复的函数实现逐项产出

items = ["a", "b"]
iterator = iter(items)
assert next(iterator) == "a"
assert next(iterator) == "b"
assert next(iterator, "结束") == "结束"

不提供默认值时，耗尽会抛 StopIteration；for 会识别它并结束遍历。常见对象通过 __iter__() 提供迭代器，由迭代器的 __next__() 产生下一项。今天认识协议即可。

同一个迭代器不会自动倒回起点：

items = [1, 2, 3]
iterator = iter(items)
assert list(iterator) == [1, 2, 3]
assert list(iterator) == []
assert list(iter(items)) == [1, 2, 3]

列表可以提供新的迭代器，原迭代器已经耗尽。为了调试而先执行 list(records)，可能把后面的处理流程需要的数据全部消费掉；JS generator 对象也有类似行为。

6. 用 yield 保留执行位置

如果每次取得下一项都需要运行逻辑，可以用生成器函数，不必手写保存状态的迭代器类：

def generate_ids():
    print("开始")
    yield "a"
    print("继续")
    yield "b"
    print("结束")

ids = generate_ids()  # 尚未执行函数体
assert next(ids) == "a"  # 打印开始
assert next(ids) == "b"  # 打印继续
assert next(ids, None) is None  # 打印结束，生成器耗尽

调用含 yield 的函数创建生成器对象；next 驱动函数执行到 yield，交出值并暂停，保留局部变量和执行位置；下一次继续。return 则终止生成器。

JS 的 function* 同样可以暂停产出，但 JS next 返回 { value, done }，Python next 直接返回值，耗尽时抛 StopIteration。

表达式也能创建生成器：(record["id"] for record in records)。与列表推导式相比，它按需计算输出；简单生成器表达式的最左侧可迭代部分会在创建时求值，不要把“惰性”理解成所有表达式都完全延迟。

7. 惰性计算的收益与边界

逐批向模型服务提交文档时，可以生成一批、处理一批，不必先保存所有输出批次。

但如果输入本身已经是巨大的列表，它占据的内存不会消失；调用方执行 list(generator) 或 sorted(generator)，也会收集数据。生成器节约的是不必预先保存全部输出的空间，收益取决于完整数据流。

还要注意错误时机：生成器函数体内的校验通常在开始迭代时执行，不是在创建对象时执行。下一节练习会验证这一点。

yield from other_iterable 表示把另一个可迭代对象的元素继续向外产出，今天识别其基本用途即可，不深入 send/throw。

输出预测

先不运行，写出每一行输出：

def values():
    print("A")
    yield 1
    print("B")
    yield 2

g = values()
print("C")
print(next(g))
print(list(g))
print(list(g))

答案为 C、A、1、B、[2]、[]。创建 g 不执行函数体，next 消费第一项，第一次 list 消费剩余项，第二次 list 面对的是已耗尽的生成器。

三个编码练习

先独立完成，再看参考答案。

1. 计时装饰器 timed：使用 perf_counter，支持位置和关键字参数，保留名称与返回值，不吞异常。基础版只记录成功调用耗时。
2. 批次生成器 batched：接受可迭代数据和正整数 size，产出列表批次；覆盖空输入、不整除和非法 size；不同批次不能共享同一个列表。
3. 闭包与迭代消耗：创建三个分别返回 0、1、2 的函数；取得迭代器第一项后，将剩余项转为列表，解释数据去向。

练习参考答案

A. 计时装饰器

from functools import wraps
from time import perf_counter

def timed(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        started = perf_counter()
        result = func(*args, **kwargs)
        print(f"{func.__name__}: {perf_counter() - started:.6f}s")
        return result
    return wrapper

@timed
def count_words(text, *, separator=None):
    return len(text.split(separator))

assert count_words("python agent") == 2
assert count_words("a,b", separator=",") == 2
assert count_words.__name__ == "count_words"

@timed
def fail():
    raise ValueError("示例错误")

try:
    fail()
except ValueError as exc:
    assert str(exc) == "示例错误"
else:
    raise AssertionError("异常不应被吞掉")

异常发生时，计时输出不会执行，但异常继续向调用方传播。如果要求失败也记录耗时，可在学习 Day 3 的 finally 后扩展；不要为了让日志打印就把错误转成成功结果。

B. 批次生成器

def batched(records, size):
    if size <= 0:
        raise ValueError("size 必须大于 0")
    batch = []
    for record in records:
        batch.append(record)
        if len(batch) == size:
            yield batch
            batch = []
    if batch:
        yield batch

assert list(batched([], 2)) == []
assert list(batched(range(5), 2)) == [[0, 1], [2, 3], [4]]
result = list(batched(range(4), 2))
assert result[0] is not result[1]

try:
    list(batched([1], 0))  # 消费时才执行函数体中的校验
except ValueError:
    pass
else:
    raise AssertionError("非法 size 应失败")

本题约定 size 输入为整数。yield 后创建新列表，不能改为 batch.clear()，否则会清空已经交给调用者的列表。练习中 list 用于小规模结果断言，真实大数据处理应直接逐批消费。

C. 闭包与消费状态

def make_callbacks():
    return [lambda value=i: value for i in range(3)]

assert [fn() for fn in make_callbacks()] == [0, 1, 2]
iterator = iter([10, 20, 30])
assert next(iterator) == 10
assert list(iterator) == [20, 30]
assert list(iterator) == []

默认参数记录每次创建函数时的对象，迭代器记录当前消费位置。它们都要求区分“定义或创建”和“之后调用”两个时刻。

面试 Q&A

下面先把每个问题的原因和运行过程讲清楚，最后提供一段可以用自己的话复述的面试回答。代码中的断言用于核对理解，不需要背代码。

1. 装饰器解决什么问题？

假设读取文档、调用模型、保存结果三个函数都需要记录日志。如果把相同日志写进每个函数，以后修改日志格式就要改多个地方，也容易漏掉某个函数。装饰器可以把共同的调用逻辑集中起来，让业务函数继续表达自己的工作。

从你熟悉的 JS 出发，它可以理解为 const wrapped = withLogging(original)：一个函数接收另一个函数，返回增加了行为的新函数。Python 的 @ 语法让这种应用关系紧挨着函数定义展示。

from functools import wraps

events = []

def logged(func):
    events.append("应用装饰器")
    @wraps(func)
    def wrapper(*args, **kwargs):
        events.append("开始调用")
        result = func(*args, **kwargs)
        events.append("调用成功")
        return result
    return wrapper

@logged
def load_document(name):
    events.append("执行业务")
    return {"name": name}

assert events == ["应用装饰器"]
assert load_document("guide.md") == {"name": "guide.md"}
assert events == ["应用装饰器", "开始调用", "执行业务", "调用成功"]

这里有两个不同的执行阶段：

1. 执行定义时：先创建原始 load_document 函数，再把它传给 logged。logged 返回 wrapper，名称 load_document 随后指向这个 wrapper。此时原始业务函数还没运行。
2. 调用时：执行 wrapper，它先记录日志，再通过闭包保留的 func 调用原函数，最后返回原函数结果。

所以“装饰器在定义时执行”不表示“每次调用前的日志在定义时打印”。前者是创建包装关系，后者是实际调用包装函数。

装饰器也不一定返回新包装器。工具注册装饰器可能把原函数存进注册表，再原样返回它。看到 @tool 或 @route，只能先判断“这里应用了装饰器”，具体行为仍要看实现。

容易误解的地方： 上面的业务函数一旦抛异常，“调用成功”就不会执行，异常继续向上传播。装饰器不会自动处理失败、重试或异步等待，这些行为需要明确编写。

面试可以这样回答：

装饰器用于集中实现函数的通用行为，例如日志、计时或注册。常见做法是接收原函数，返回包装函数，包装函数再调用原函数。@ 语法在函数定义执行时应用装饰器，而包装器中的调用逻辑通常在之后调用函数时执行。我会区分注册型和包装型装饰器，也会确认参数、返回值和异常有没有正确传递。

2. 为什么包装函数通常要使用 wraps？

包装器替换了原函数的外部名称，但包装器也是一个真实函数，有它自己的名字和文档。如果什么都不处理，外部看到的可能是 wrapper，而不再是原来的业务函数信息。

from functools import wraps

def plain_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

def careful_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

def summarize(text):
    """返回文本前五个字符。"""
    return text[:5]

plain = plain_decorator(summarize)
careful = careful_decorator(summarize)
assert plain.__name__ == "wrapper"
assert careful.__name__ == "summarize"
assert careful.__doc__ == summarize.__doc__
assert careful.__wrapped__ is summarize
assert careful("abcdef") == "abcde"

wraps(func) 帮助复制函数名称、文档等元信息，并设置 __wrapped__ 指向原函数。这样日志显示业务名称，文档工具可以取得原说明，支持展开包装关系的检查工具也可以找到原函数。

可以把元信息理解为函数的“说明资料”，把函数体理解为真正执行的代码。wraps 主要维护说明资料和原函数关联，不是把 wrapper 的代码替换回原函数。

容易误解的地方： wrapper 仍然使用 *args, **kwargs 接收参数。某些检查工具可以沿 __wrapped__ 展示原函数签名，但 wraps 并没有重写包装器的实际参数列表，也不执行类型校验。同步 wrapper 包装 async 函数时，仍需自己考虑等待、计时和异常边界，wraps 不会替它执行 await。

面试可以这样回答：

装饰后，调用者拿到的通常是包装函数，原函数的名字和文档容易丢失。functools.wraps 用来保留这些元信息，并通过 wrapped 关联原函数，便于调试、生成文档和函数检查。它不负责业务逻辑，也不会自动处理异步调用或参数校验。

3. iterable、iterator、generator 有什么关系？

可以从“数据来源”和“当前读到哪里”区分。列表是一个可以遍历的数据来源；从列表创建的迭代器则记录这一次遍历已经走到哪个位置。生成器是实现迭代器的一种方式，用函数的暂停与恢复保存进度。

概念	你需要知道什么	常见例子
iterable，可迭代对象	能交给 iter() 获取迭代器	list、str、dict
iterator，迭代器	能交给 next() 取得下一项，并维护进度	iter(list) 的结果
generator，生成器	是一种迭代器，由生成器函数或生成器表达式创建	含 yield 的函数返回的对象

items = [10, 20]
left = iter(items)
right = iter(items)

assert next(left) == 10
assert next(left) == 20
assert next(right) == 10  # right 有自己的遍历进度
assert iter(left) is left
assert next(left, "结束") == "结束"

def produce():
    yield 10
    yield 20

generator = produce()
assert iter(generator) is generator
assert list(generator) == [10, 20]
assert list(generator) == []

迭代器本身也可迭代，通常 iter(iterator) 返回它自己，所以再次对同一个迭代器执行 for 或 list，不会自动重新开始。列表则能够创建新的迭代器，支持多次从头遍历。

for 不需要知道数据来自文件还是列表，它只需要取得迭代器，不断请求下一项，并在 StopIteration 时结束。这就是迭代协议的价值：不同来源的数据可以交给相同的消费代码。

与 JS 对照： Python 的 __iter__ 可联系 JS 的 [Symbol.iterator]；两边迭代器都提供“下一项”，但 JS next 返回 {value, done}，Python next 返回元素，并以 StopIteration 表示耗尽。

容易误解的地方： 可迭代不等于可重复读取，也不等于支持下标或 len。生成器是迭代器，但不能由此推断它和列表拥有同样的操作。

面试可以这样回答：

iterable 表示可以取得迭代器的对象，iterator 表示能逐项提供值并维护遍历状态的对象。generator 是一种迭代器，用 yield 保存和恢复执行进度。列表可以创建多个独立迭代器，而同一个生成器被消耗后不会自动重置。for 依赖的是迭代协议，所以能够统一处理多种数据来源。

4. 调用生成器函数会执行函数体吗？

不会立即执行函数体。调用时会创建生成器对象，等 next、for、list 等操作请求数据时，才开始运行函数体。执行到 yield 后暂停，下次请求继续运行。

events = []

def produce():
    events.append("开始")
    yield "a"
    events.append("恢复")
    yield "b"
    events.append("结束")

g = produce()
assert events == []
assert next(g) == "a"
assert events == ["开始"]
assert next(g) == "b"
assert events == ["开始", "恢复"]
assert next(g, None) is None
assert events == ["开始", "恢复", "结束"]

第一次 next 执行到第一个 yield，第二次 next 从那里继续，第三次才运行最后的“结束”并耗尽。拿到最后一个产出值时，不代表 yield 后面的代码已经执行。

这会影响错误发生的位置：

def produce_batches(size):
    if size <= 0:
        raise ValueError("size 必须大于 0")
    yield size

g = produce_batches(0)  # 函数体内的校验还未发生
try:
    next(g)
except ValueError:
    pass
else:
    raise AssertionError("消费时应该触发校验")

如果只在 g = produce_batches(0) 外层捕获异常，可能捕获不到消费阶段抛出的错误。需要把异常处理放在真正遍历的位置，或者用普通外层函数先校验，再返回内部生成器。

容易误解的地方： “不执行函数体”不表示调用时什么都不求值。调用表达式中的实参会先求值；函数参数绑定错误，例如缺少必填参数，也会在调用时暴露。生成器表达式还存在最左侧可迭代部分的求值边界，不应把所有形式概括为完全延迟。

面试可以这样回答：

调用生成器函数只创建生成器对象，函数体通常在消费时开始执行。yield 交出值并保留局部状态，下一次消费从暂停处继续。这也意味着函数体中的校验和异常可能延迟到遍历时发生，所以我会检查异常处理是否覆盖实际消费过程。

5. 生成器为什么可能节省内存？

假设有十万条文档 ID。列表方案会先保存十万条输出，生成器方案可以在消费者需要时只计算下一条。

# 只用少量数据演示输出行为，不把它当性能测试。
eager = [f"doc-{i}" for i in range(4)]
lazy = (f"doc-{i}" for i in range(4))
assert eager == ["doc-0", "doc-1", "doc-2", "doc-3"]
assert next(lazy) == "doc-0"
assert list(lazy) == ["doc-1", "doc-2", "doc-3"]

列表预先保存全部输出元素；生成器主要保存继续执行所需的状态，并按需求产出。对“读取一条 → 转换一条 → 写出一条”的流水线，能够避免积累全部中间结果。

但必须看完整的数据流：

• 输入如果已是大列表，该列表仍然占内存，生成器不会删除它。
• list(generator) 会收集全部输出，sorted(generator) 也需要收集数据再排序。
• 生成器内部如果维护不断增长的缓存，同样可能大量占用内存。
• 分批生成会保留一批数据，而不是完全没有内存开销；已产出的批次如果被消费者全部保存，总内存仍会增长。

例如每次给模型传 32 条文档，批次生成器可以只组织当前批次；如果调用方先把所有批次存入一个列表，就失去了这部分按需处理的收益。

容易误解的地方： 惰性不保证更快。生成器的主要优势是按需计算和减少不必要的结果保存，实际耗时取决于业务逻辑和调用方式。也不能只用生成器对象自身的大小代表整条数据流占用的内存。

面试可以这样回答：

生成器通过按需产出，避免预先构造和保存全部输出，因此适合大文件、文档流水线和批处理。但内存收益不是自动成立的，还要看输入是否已经物化、生成器内部保留什么状态，以及消费者是否把结果全部收集。我会评估整条数据流，而不只看有没有 yield。

6. 闭包晚绑定是什么？为什么循环里的函数都返回最后一个值？

闭包保留的是访问外层名称绑定的能力，不一定是创建函数那一刻的值副本。函数真正执行时，再读取该名称当前绑定的对象，这就是常说的晚绑定。

def build_callbacks():
    callbacks = []
    for i in range(3):
        callbacks.append(lambda: i)
    return callbacks

callbacks = build_callbacks()
assert [fn() for fn in callbacks] == [2, 2, 2]

时间顺序是：

1. 循环把 i 依次绑定到 0、1、2，每轮创建一个函数，但没有调用它。
2. 三个函数访问的是同一外层 i 的绑定。
3. 循环结束后，i 为 2；这时调用三个函数，它们都读到 2。

它不是 lambda 独有的问题，改写成内部 def 而不改变绑定关系，也会出现同样行为。JS 使用 for (let i = ...) 时每轮有独立绑定；Python 普通 for 不提供这一对应行为。

有两种常见修复：

# 方法一：定义时求值默认参数，保存本轮整数。
def with_defaults():
    return [lambda value=i: value for i in range(3)]

# 方法二：每次工厂调用都有独立的局部绑定。
def make_callback(value):
    def callback():
        return value
    return callback

assert [fn() for fn in with_defaults()] == [0, 1, 2]
assert [fn() for fn in [make_callback(i) for i in range(3)]] == [0, 1, 2]

第一种利用 Day 1 的默认参数求值规则，第二种利用不同函数调用的独立局部作用域。两者都不是自动深拷贝：

items = []
read_items = lambda saved=items: saved
items.append("python")
assert read_items() == ["python"]

默认参数保存了列表对象，后面修改这个列表，仍能观察到变化。因此“固定对象绑定”与“复制对象内容”必须分开。

面试可以这样回答：

闭包晚绑定指内部函数执行时读取外层名称的绑定，而不是默认保存创建时的值。循环中延迟调用的多个函数可能共享同一个循环变量，因此都得到最后的值。可以用默认参数在定义时保存本轮对象，或者用工厂函数创建独立作用域；如果保存的是可变对象，还要另外考虑是否需要复制。

今日验收

能独立写出上述三个练习、解释输出顺序，并指出 batch.clear() 和重复消费迭代器的问题，就完成今天的主要目标。下一步学习异常、资源管理与模块入口，让包装与生成的数据流在失败时也有明确行为。