编程进阶网
杨充
2019-03-28
目录

函数高级特性剖析：装饰器 / 生成器 / 上下文管理器

# 第 11 章 函数高级特性剖析

# 目录介绍

这三种特性是 Python 面试高频考点，也是写出优雅代码的"大杀器"。

# 11.1 装饰器深度剖析

Python 中一切都是对象——函数也不例外。函数可以被赋值给变量、作为参数传递、作为返回值返回。装饰器的本质是接受一个函数、返回一个新函数的高阶函数。

# 装饰器本质 = 高阶函数的语法糖
@decorator
def target():
    pass

# 等价于：
def target():
    pass
target = decorator(target)   # 函数被替换为装饰后的版本
1
2
3
4
5
6
7
8
9

# 11.1.1 闭包与装饰器基础

闭包 = 函数内部定义的函数，能记住外层函数的作用域（即使外层函数已返回）：

def outer(x):
    def inner(y):
        return x + y          # inner 记住了 outer 的 x
    return inner

add_5 = outer(5)              # outer(5) 返回 inner 函数
print(add_5(3))               # 8——inner 仍然能访问 x=5
print(add_5.__closure__)      # (<cell: int object at ...>,)
print(add_5.__closure__[0].cell_contents)  # 5——闭包捕获的值
1
2
3
4
5
6
7
8
9

# 11.1.2 无参装饰器

#!/usr/bin/env python3
import functools, time

def timer(func):
    """计时器——统计函数执行时间"""
    @functools.wraps(func)        # 保留原函数的 __name__ / __doc__
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"[timer] {func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_api(n):
    time.sleep(n / 100)
    return n
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

为什么要 @functools.wraps？ 装饰器用 wrapper 替换了原函数——如果没有 wraps，slow_api.__name__ 会变成 "wrapper"。这在调试、文档生成、序列化场景下会引发隐蔽的 bug。

# 11.1.3 带参装饰器

带参数装饰器的本质是"装饰器工厂"——先调用工厂拿到真正的装饰器，再用装饰器包装函数：

#!/usr/bin/env python3
import functools, time, logging

def retry(max_retries=3, delay=1, exceptions=(Exception,)):
    """重试装饰器工厂"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_retries: raise
                    logging.warning(
                        f"[retry] {func.__name__} 第 {attempt} 次失败: {e}，"
                        f" {delay}s 后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_retries=5, delay=2, exceptions=(TimeoutError, ConnectionError))
def call_api(url): pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 11.1.4 类装饰器

函数装饰器三层嵌套有时难读——类装饰器用 __init__ + __call__ 让逻辑更清晰：

#!/usr/bin/env python3
import functools, time

class RateLimiter:
    """限流装饰器——类版本更易维护"""
    def __init__(self, calls_per_second=10):
        self.rate = calls_per_second
        self.tokens = calls_per_second
        self.last_check = time.monotonic()

    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            now = time.monotonic()
            elapsed = now - self.last_check
            self.tokens = min(self.rate, self.tokens + elapsed * self.rate)
            self.last_check = now
            if self.tokens < 1:
                time.sleep((1 - self.tokens) / self.rate)
                self.tokens = 0
            else:
                self.tokens -= 1
            return func(*args, **kwargs)
        return wrapper

@RateLimiter(calls_per_second=5)
def api_call(): pass
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 11.1.5 实战：缓存与权限

#!/usr/bin/env python3
import functools, hashlib, json

# ---- 1. 内存缓存装饰器 ----
def memoize(func):
    cache = {}
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        key = hashlib.md5(json.dumps((args, kwargs), sort_keys=True).encode()).hexdigest()
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        return cache[key]
    return wrapper

# ---- 2. 权限校验装饰器 ----
def require_role(role):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(user, *a, **kw):
            if getattr(user, 'role', None) != role:
                raise PermissionError(f"需要 {role} 权限")
            return func(user, *a, **kw)
        return wrapper
    return decorator

# ---- 3. 插件注册器 ----
_plugins = {}
def register(name):
    def decorator(func):
        _plugins[name] = func
        @functools.wraps(func)
        def wrapper(*a, **kw): return func(*a, **kw)
        return wrapper
    return decorator
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 11.2 迭代器与生成器

迭代器协议：实现 __iter__() 和 __next__()（耗尽抛出 StopIteration）。生成器 = 用函数语法写出的迭代器——yield 把普通函数变成状态机。

# 11.2.1 迭代协议源码视角

class CountDown:
    def __init__(self, start): self.current = start
    def __iter__(self): return self
    def __next__(self):
        if self.current <= 0: raise StopIteration
        self.current -= 1; return self.current + 1

# for i in CountDown(3): 等价于——
it = iter(CountDown(3))
while True:
    try: i = next(it); print(i)
    except StopIteration: break
1
2
3
4
5
6
7
8
9
10
11
12

# 11.2.2 yield 状态机

def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield b; a, b = b, a + b

for val in fibonacci(10):
    print(val, end=' ')           # 1 1 2 3 5 8 13 21 34 55

# 生成器表达式——比列表推导省内存
square_list = [x**2 for x in range(10**6)]       # ~8MB
square_gen  = (x**2 for x in range(10**6))       # < 1KB
1
2
3
4
5
6
7
8
9
10
11

# 11.2.3 双向通信机制

def accumulator():
    total = 0
    while True:
        value = yield total        # 右边输出，左边接收 send 的值
        if value is None: break
        total += value
    return total

gen = accumulator()
next(gen)                          # 必须启动！
print(gen.send(10))                # 10
print(gen.send(20))                # 30
try:
    gen.send(None)
except StopIteration as e:
    print(f"最终: {e.value}")      # 25
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 11.2.4 yield from 委托

def flatten(nested):
    for sublist in nested:
        yield from sublist         # 等价于 for item in sublist: yield item

print(list(flatten([[1,2], [3,4], [5]])))  # [1,2,3,4,5]
1
2
3
4
5

yield from 还透明传递 send()/throw()/close() 调用，是协程嵌套的基础。

# 11.2.5 大文件流式管道

import re

def read_lines(filename):
    with open(filename) as f:
        for line in f: yield line.strip()

def filter_errors(lines):
    for line in lines:
        if 'ERROR' in line: yield line

def extract_ips(lines):
    for line in lines:
        m = re.search(r'\d+\.\d+\.\d+\.\d+', line)
        if m: yield m.group()

# 管道——每个阶段是生成器，全程惰性
pipeline = extract_ips(filter_errors(read_lines('huge.log')))
for ip in pipeline: print(ip)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 11.3 上下文管理器

with 协议：__enter__ 进入时调用，__exit__ 无论异常与否都会调用——这保证了资源释放。

# 11.3.1 手动实现方法

class DatabaseConnection:
    def __init__(self, db_path): self.db_path = db_path
    def __enter__(self): self.conn = sqlite3.connect(self.db_path); return self.conn
    def __exit__(self, *args): self.conn.close()
1
2
3
4

# 11.3.2 contextlib 捷径

import contextlib, time

@contextlib.contextmanager
def timer(name=''):
    start = time.perf_counter()
    yield                                 # yield 前 = __enter__, 后 = __exit__
    print(f"[{name}] {time.perf_counter() - start:.3f}s")

@contextlib.contextmanager
def chdir(path):
    import os; old = os.getcwd(); os.chdir(path)
    try: yield
    finally: os.chdir(old)
1
2
3
4
5
6
7
8
9
10
11
12
13

# 11.3.3 ExitStack 嵌套

# 一行打开多个文件（Python 3.3+）
with open('a.txt') as f1, open('b.txt') as f2:
    for l1, l2 in zip(f1, f2): pass

# ExitStack——运行时动态管理
from contextlib import ExitStack
with ExitStack() as stack:
    handles = [stack.enter_context(open(f)) for f in ['a.txt','b.txt','c.txt']]
1
2
3
4
5
6
7
8

# 11.3.4 事务与文件锁

import contextlib, fcntl, sys

# ---- 数据库事务——自动 commit/rollback ----
@contextlib.contextmanager
def transaction(conn):
    try: yield conn; conn.commit()
    except Exception: conn.rollback(); raise

# ---- 文件锁——防并发写冲突 ----
@contextlib.contextmanager
def file_lock(filepath):
    with open(filepath, 'a') as f:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try: yield f
        finally: fcntl.flock(f.fileno(), fcntl.LOCK_UN)

# ---- 捕获 print 输出 ----
@contextlib.contextmanager
def capture_stdout():
    from io import StringIO
    old = sys.stdout; sys.stdout = buf = StringIO()
    try: yield buf
    finally: sys.stdout = old
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 11.4 三种特性内在联系

它们共享同一个底层机制：Python 的协议 + 语法糖。

特性 协议 语法糖 核心作用
装饰器 无（闭包+高阶函数） @decorator 函数的增强包装
生成器 __iter__/__next__ yield / yield from 惰性计算+状态保持
上下文管理器 __enter__/__exit__ with 资源生命期确定性管理

理解协议，就理解了 Python 的核心设计哲学。

#Python#原理
上次更新: 2026/06/17, 12:47:39
部署与并发实战
并发底层原理揭秘

← 并发底层原理揭秘→

最近更新
01
信号崩溃快速排查
06-15
02
CoreDump破案
06-15
03
perf火焰图实战
06-15
更多文章>
Theme by Vdoing | Copyright © 2019-2026 杨充 | MIT License | 桂ICP备2024034950号 | 桂公网安备45142202000030