装饰器与上下文管理器

装饰器给可调用对象披上一件外衣

函数是一个可调用对象：也就是可以使用 () 调用的对象。装饰器是普通的 Python 代码，它接收一个可调用对象，并返回供人们使用的可调用对象。

def announce(function):
    def wrapper(*args, **kwargs):
        print("before")
        result = function(*args, **kwargs)
        print("after")
        return result

    return wrapper


@announce
def greeting(name):
    return f"Hello, {name}"


print(greeting("Mina"))

输出：

before
after
Hello, Mina

@announce 表示 greeting = announce(greeting)。Python 执行函数定义时会完成装饰；之后每次调用时，包装器才会运行。一个透明的包装器会转发 *args 和 **kwargs，并返回原始结果。如果忘记 return result，调用方会在不知不觉中得到 None。

保留名称并配置这件外衣

普通包装器的名称是 wrapper，而不是 greeting。functools.wraps() 会保留原始名称、文档、注解以及 __wrapped__ 链接。

可配置的装饰器是一个装饰器工厂，它包含三个小层次：

from functools import wraps


def tagged(label):                       # receive configuration
    if not label:
        raise ValueError("label must not be empty")

    def decorate(function):              # receive the callable
        @wraps(function)
        def wrapper(*args, **kwargs):     # receive call arguments
            return f"[{label}] {function(*args, **kwargs)}"

        return wrapper

    return decorate


@tagged("study")
def total(a, b):
    """Add two study times."""
    return a + b


print(total(20, 15))
print(total.__name__)

输出：

[study] 35
total

tagged("study") 和 decorate(total) 会在定义时运行。最内层的包装器会在调用时运行。如果叠加多个装饰器，最靠近函数的装饰器会最先应用；只有当你能够清楚解释它们的顺序时，才应该使用装饰器堆叠。

with 打开并关闭一道门

上下文管理器包围的是一个代码块，而不是一次调用。with 会调用 __enter__()，运行代码块，然后始终调用 __exit__()。

class Notebook:
    def __enter__(self):
        print("open")
        self.notes = []
        return self.notes

    def __exit__(self, exc_type, exc_value, traceback):
        print("close")
        return False


with Notebook() as notes:
    notes.append("context managers")
    print(notes)

输出：

open
['context managers']
close

__enter__() 返回的值会成为 notes。执行成功时，传给 __exit__() 的三个异常参数都是 None；执行失败时，它们会描述发生的异常。返回真值会抑制该异常。返回 False 或 None 则会让异常继续传播。抑制异常必须是范围明确且有文档说明的选择，绝不能是意外返回的 True。

contextlib 让一道门变得更简洁

对于只有一条简单进入和退出路径的情况，contextlib.contextmanager 可以把生成器变成上下文管理器：

from contextlib import contextmanager


@contextmanager
def section(name):
    print("open", name)
    try:
        yield []
    finally:
        print("close", name)


with section("testing") as notes:
    notes.append("arrange-act-assert")
    print(notes)

输出：

open testing
['arrange-act-assert']
close testing

yield 之前的代码负责进入；产出的值属于这个代码块；finally 保证清理工作一定执行。如果代码块执行失败，其异常会到达 yield，并且通常会在清理完成后继续传播。只有当管理器具有明确的审计或回滚规则时，才应该宽泛地捕获异常，之后使用不带参数的 raise 来保留原始回溯信息。

保持边界精简：只关闭管理器自己拥有的资源；优先使用资源原生支持的 with；每次使用 with 时，都要创建一个新的基于生成器的管理器，因为每个实例只能使用一次。

构建具有事务功能的阅读日志

将这个完整的标准库项目保存为 reading_journal.py：

from contextlib import contextmanager
from dataclasses import dataclass, field
from functools import wraps


@dataclass
class Journal:
    entries: list[str] = field(default_factory=list)
    audit: list[str] = field(default_factory=list)


def audited(audit, label):
    def decorate(function):
        @wraps(function)
        def wrapper(*args, **kwargs):
            audit.append(f"{label}:start")
            try:
                result = function(*args, **kwargs)
            except Exception as error:
                audit.append(f"{label}:error:{type(error).__name__}")
                raise
            else:
                audit.append(f"{label}:ok")
                return result

        return wrapper

    return decorate


@contextmanager
def staged(journal):
    pending = []
    journal.audit.append("batch:open")
    try:
        yield pending
    except Exception:
        journal.audit.append("batch:rollback")
        raise
    else:
        journal.entries.extend(pending)
        journal.audit.append(f"batch:commit:{len(pending)}")
    finally:
        journal.audit.append("batch:close")


def parse_entry(line):
    topic, separator, minutes_text = line.partition("|")
    if not separator or not topic.strip():
        raise ValueError("expected topic|minutes")
    try:
        minutes = int(minutes_text)
    except ValueError as cause:
        raise ValueError("minutes must be an integer") from cause
    if minutes <= 0:
        raise ValueError("minutes must be positive")
    return f"{topic.strip()} ({minutes} min)"


def make_importer(journal):
    @audited(journal.audit, "import")
    def import_lines(lines):
        with staged(journal) as pending:
            for line in lines:
                pending.append(parse_entry(line))
        return len(pending)

    return import_lines


journal = Journal()
import_lines = make_importer(journal)

count = import_lines(["Decorators | 25", "Contexts | 30"])
saved = journal.entries.copy()
print("Imported:", count)

try:
    import_lines(["Valid | 10", "Broken | many"])
except ValueError as error:
    print("Rejected:", error)

assert journal.entries == saved
assert import_lines.__name__ == "import_lines"
print("Entries:", journal.entries)
print("Last audit:", journal.audit[-5:])

运行 python3 reading_journal.py：

Imported: 2
Rejected: minutes must be an integer
Entries: ['Decorators (25 min)', 'Contexts (30 min)']
Last audit: ['import:start', 'batch:open', 'batch:rollback', 'batch:close', 'import:error:ValueError']

装饰器负责调用层面的审计事件。上下文管理器负责待处理数据、提交、回滚和清理。宽泛捕获异常的代码各自只有一项有文档说明的任务，并且会立即重新抛出异常；失败的批次无法改变已有条目。

三个小任务

1. 装饰一个带有文档字符串的函数，并验证它的结果、__name__ 和 __doc__。
2. 让 @repeat(times) 在装饰期间拒绝零，并让一个函数恰好调用三次。
3. 修改 Notebook.__exit__()，使它只抑制 KeyError，然后证明 ValueError 仍然会传播出去。

准备好学习第 19 章了吗？

• 我知道装饰器是一种普通的可调用对象转换。
• 我的包装器会转发参数、返回结果，并使用 @wraps。
• 我能够解释装饰器工厂的三个层次。
• 我知道 __enter__() 返回什么，以及 __exit__() 何时运行。
• 我会把无条件执行的清理工作放进 finally，并有意识地决定是否抑制异常。
• 我运行了阅读日志，并证明失败的批次会回滚。

接下来，你将使用测试、调试器以及经过谨慎配置的日志，收集三类证据。