python

Python 装饰器与上下文管理器：从语法糖到工程实践

Python 装饰器与上下文管理器完全指南：从基础语法到类装饰器、带参数装饰器、装饰器链；上下文管理器的协议实现与 contextlib 工具；以及 LRU 缓存、重试机制、性能计时等 6 个企业级装饰器实战。

spcent 2026-08-13 8 分钟阅读 3753 字

装饰器和上下文管理器是 Python 的两大"魔法特性"。它们让代码更优雅、更易维护，但也常常让初学者困惑。本文把它们拆解为"可理解的积木"，让你不仅会用，还能自己写。

1. 装饰器本质：函数 wrapper

装饰器本质上是一个接收函数作为参数并返回新函数的高阶函数。

# 最简装饰器：什么都不做
def my_decorator(func):
    def wrapper():
        print("函数调用前")
        func()
        print("函数调用后")
    return wrapper

def say_hello():
    print("Hello!")

# 不用 @ 语法糖
say_hello = my_decorator(say_hello)
say_hello()
# 输出：
# 函数调用前
# Hello!
# 函数调用后

核心理解：装饰器 = 接收函数 → 包装新功能 → 返回包装后的函数。

2. 基础装饰器：手写与 @语法糖

2.1 @语法糖

@my_decorator
def say_hello():
    print("Hello!")

# 等价于：
# say_hello = my_decorator(say_hello)

2.2 保留原函数信息（重要！）

import functools

def my_decorator(func):
    @functools.wraps(func)   # ✅ 关键：保留原函数的 __name__, __doc__
    def wrapper(*args, **kwargs):
        """wrapper doc"""
        print("调用前")
        result = func(*args, **kwargs)
        print("调用后")
        return result
    return wrapper

@my_decorator
def greet(name):
    """向某人打招呼"""
    print(f"Hi, {name}")

print(greet.__name__)    # greet（因为有 @wraps）
print(greet.__doc__)     # 向某人打招呼

没有 @wraps：greet.__name__ 会变成 'wrapper'，help(greet) 会显示 wrapper 的文档。这在调试时非常痛苦。

2.3 处理函数参数和返回值

def log_call(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"调用 {func.__name__}({args}, {kwargs})")
        result = func(*args, **kwargs)
        print(f"返回: {result}")
        return result
    return wrapper

@log_call
def add(a, b):
    return a + b

add(3, 5)
# 输出：
# 调用 add((3, 5), {})
# 返回: 8

3. 带参数的装饰器

3.1 三层嵌套结构

def repeat(times):
    """让函数执行多次"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for _ in range(times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(times=3)
def greet(name):
    print(f"Hello, {name}!")

greet("Alice")
# Hello, Alice!
# Hello, Alice!
# Hello, Alice!

结构解析：

repeat(times=3)
  → 返回 decorator
    → decorator(greet)
      → 返回 wrapper
        → wrapper() 执行 3 次

3.2 带可选参数的装饰器

def smart_decorator(arg=None):
    """既可以 @decorator 也可以 @decorator(arg)"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if arg:
                print(f"参数: {arg}")
            return func(*args, **kwargs)
        return wrapper
    
    if callable(arg):
        return decorator(arg)
    return decorator

# 两种方式都可以
@smart_decorator           # 无参数
@smart_decorator("debug")  # 有参数
def my_func():
    pass

4. 类装饰器

类装饰器 = 用类来实现装饰器功能。

4.1 用类实现装饰器

class CountCalls:
    """统计函数被调用次数"""
    
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self.func = func
        self.count = 0
    
    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"{self.func.__name__} 被调用了 {self.count} 次")
        return self.func(*args, **kwargs)

@CountCalls
def say_hello():
    print("Hello!")

say_hello()
say_hello()
say_hello()
# → say_hello 被调用了 1 次
# → say_hello 被调用了 2 次
# → say_hello 被调用了 3 次

4.2 带参数的类装饰器

class RateLimiter:
    """限流装饰器：限制函数调用频率"""
    
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
    
    def __call__(self, func):
        functools.update_wrapper(self, func)
        
        import time
        calls = []
        
        def wrapper(*args, **kwargs):
            now = time.time()
            # 清理过期记录
            calls[:] = [c for c in calls if now - c < self.period]
            
            if len(calls) >= self.max_calls:
                raise RuntimeError(f"限流：{self.period} 秒内最多 {self.max_calls} 次调用")
            
            calls.append(now)
            return func(*args, **kwargs)
        
        return wrapper

@RateLimiter(max_calls=5, period=60)
def api_call():
    print("API 调用成功")

5. 多个装饰器叠加

装饰器从下往上执行：

@decorator_a
@decorator_b
@decorator_c
def func():
    pass

# 等价于：
# func = decorator_a(decorator_b(decorator_c(func)))

执行顺序示例

def decorator_a(func):
    @functools.wraps(func)
    def wrapper():
        print("A - before")
        func()
        print("A - after")
    return wrapper

def decorator_b(func):
    @functools.wraps(func)
    def wrapper():
        print("B - before")
        func()
        print("B - after")
    return wrapper

@decorator_a
@decorator_b
def say_hi():
    print("Hi!")

say_hi()
# 输出：
# A - before
# B - before
# Hi!
# B - after
# A - after

6. 企业级装饰器实战

6.1 性能计时

import functools
import time
from typing import Callable

def timer(func: Callable) -> Callable:
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"⏱ {func.__name__} 耗时: {elapsed:.4f}s")
        return result
    return wrapper

# Profiling 版本：记录统计信息
class Profiler:
    def __init__(self):
        self.stats = {}
    
    def __call__(self, func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = func(*args, **kwargs)
            elapsed = time.perf_counter() - start
            
            if func.__name__ not in self.stats:
                self.stats[func.__name__] = {"calls": 0, "total": 0}
            self.stats[func.__name__]["calls"] += 1
            self.stats[func.__name__]["total"] += elapsed
            
            return result
        return wrapper
    
    def report(self):
        print("\n📊 性能报告：")
        for name, stat in sorted(self.stats.items()):
            avg = stat["total"] / stat["calls"]
            print(f"  {name}: {stat['calls']} 次, 平均 {avg:.4f}s")

profiler = Profiler()

@profiler
def slow_function():
    time.sleep(0.1)

6.2 重试机制

import time
import functools
from typing import Type

def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    """失败时自动重试"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts:
                        wait = delay * (2 ** (attempt - 1))  # 指数退避
                        print(f"⚠️ 第 {attempt} 次失败: {e}，{wait:.1f}s 后重试...")
                        time.sleep(wait)
            raise last_exception
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2.0, exceptions=(ConnectionError,))
def fetch_data():
    import random
    if random.random() < 0.7:
        raise ConnectionError("网络错误")
    return "数据"

6.3 缓存（LRU）

Python 内置的 functools.lru_cache 是最常用的装饰器之一：

from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# 性能对比
import time

# 不用缓存
# fib(35) → 约 3 秒

# 用缓存
start = time.perf_counter()
print(fibonacci(300))
print(f"耗时: {time.perf_counter() - start:.6f}s")

# 清除缓存
fibonacci.cache_clear()

# 查看缓存信息
print(fibonacci.cache_info())
# CacheInfo(hits=2, misses=301, maxsize=128, currsize=128)

什么时候用 LRU Cache：

纯函数（相同输入 → 相同输出）
计算成本高
输入空间有限

6.4 日志与审计

import functools
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

def audit_log(action: str):
    """记录关键操作"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            user = kwargs.get('user_id', 'anonymous')
            logger.info(f"[{datetime.now()}] {action} by user={user}: {func.__name__}")
            try:
                result = func(*args, **kwargs)
                logger.info(f"[{action}] 成功")
                return result
            except Exception as e:
                logger.error(f"[{action}] 失败: {e}")
                raise
        return wrapper
    return decorator

@audit_log("删除用户")
def delete_user(user_id: int, admin_id: int):
    logger.info(f"执行用户删除: {user_id}")

6.5 参数验证

from functools import wraps
from typing import TypeVar

T = TypeVar('T')

def validate_types(**kwargs):
    """参数类型检查装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **func_kwargs):
            for arg_name, expected_type in kwargs.items():
                if arg_name in func_kwargs:
                    value = func_kwargs[arg_name]
                    if not isinstance(value, expected_type):
                        raise TypeError(
                            f"{func.__name__}: {arg_name} 应为 {expected_type.__name__}，"
                            f"实际为 {type(value).__name__}"
                        )
            return func(*args, **func_kwargs)
        return wrapper
    return decorator

@validate_types(name=str, age=int)
def create_user(name, age):
    return {"name": name, "age": age}

create_user(name="Alice", age=25)     # ✅
# create_user(name="Bob", age="25")   # ❌ TypeError

6.6 单例模式

def singleton(cls):
    """确保类只有一个实例"""
    instances = {}
    @functools.wraps(cls)
    def wrapper(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return wrapper

@singleton
class Database:
    def __init__(self, dsn):
        print("初始化数据库连接...")
        self.dsn = dsn

db1 = Database("postgresql://localhost/db")
db2 = Database("postgresql://localhost/db")
print(db1 is db2)   # True（同一个对象）

7. 上下文管理器：with 语句的背后

7.1 为什么要用 with

# ❌ 传统写法：容易忘记关闭
f = open("data.txt", "w")
f.write("hello")
f.close()   # 如果上面报错，这里不会执行！

# ✅ with 写法：自动管理资源
with open("data.txt", "w") as f:
    f.write("hello")
# 无论是否异常，f 都会被关闭

7.2 上下文管理器协议

实现 __enter__ 和 __exit__ 两个方法：

class ManagedFile:
    """自定义文件上下文管理器"""
    
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None
    
    def __enter__(self):
        print(f"打开文件: {self.filename}")
        self.file = open(self.filename, self.mode)
        return self.file   # 返回的资源赋给 as 变量
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        exc_type: 异常类型
        exc_val: 异常值
        exc_tb: 异常追踪信息
        """
        if self.file:
            print(f"关闭文件: {self.filename}")
            self.file.close()
        
        # 返回 True 会吞掉异常，通常不应该！
        return False

# 使用
with ManagedFile("test.txt", "w") as f:
    f.write("Hello, World!")

7.3 数据库连接上下文管理器

import sqlite3

class DatabaseConnection:
    """自动管理数据库连接的事务"""
    
    def __init__(self, db_path):
        self.db_path = db_path
        self.conn = None
    
    def __enter__(self):
        self.conn = sqlite3.connect(self.db_path)
        return self.conn
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is None:
            self.conn.commit()    # 没异常则提交
            print("✅ 事务已提交")
        else:
            self.conn.rollback()  # 有异常则回滚
            print(f"❌ 事务已回滚: {exc_val}")
        self.conn.close()

# 使用
with DatabaseConnection("app.db") as conn:
    conn.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
    # conn.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))
    # 如果这里出错，自动回滚！

8. contextlib：用生成器写上下文管理器

contextlib.contextmanager 让你用更简洁的生成器语法写上下文管理器。

8.1 基础用法

from contextlib import contextmanager

@contextmanager
def managed_file(filename, mode="r"):
    """用生成器实现上下文管理器"""
    print(f"打开: {filename}")
    f = open(filename, mode)
    try:
        yield f        # yield 的值传给 as 变量
    finally:
        print(f"关闭: {filename}")
        f.close()

# 使用
with managed_file("test.txt", "w") as f:
    f.write("Hello")

执行流程：

进入 with → 执行 yield 前的代码 → yield f → 执行 with 块 → finally 块

8.2 常用 contextlib 工具

from contextlib import contextmanager, suppress, redirect_stdout
import sys

# suppress：忽略指定异常
from pathlib import Path

with suppress(FileNotFoundError):
    content = Path("maybe_not_exists.txt").read_text()

# redirect_stdout：重定向输出
with open("output.txt", "w") as f:
    with redirect_stdout(f):
        print("这会被写入文件！")

# ExitStack：动态管理多个上下文
from contextlib import ExitStack

with ExitStack() as stack:
    files = [stack.enter_context(open(fname)) for fname in ['a.txt', 'b.txt', 'c.txt']]
    # 所有文件退出时自动关闭

# closing：确保有 close() 方法的对象被关闭
from contextlib import closing
import urllib.request

with closing(urllib.request.urlopen('https://example.com')) as page:
    html = page.read()

8.3 临时修改环境变量

import os
from contextlib import contextmanager

@contextmanager
def temp_env_var(key, value):
    """临时修改环境变量，退出时恢复"""
    old_value = os.environ.get(key)
    os.environ[key] = value
    try:
        yield
    finally:
        if old_value is None:
            os.environ.pop(key, None)
        else:
            os.environ[key] = old_value

# 使用
with temp_env_var("DEBUG", "1"):
    print(os.environ["DEBUG"])   # 1
print(os.environ.get("DEBUG"))   # 恢复为原来的值

8.4 上下文管理器装饰器模式

from contextlib import contextmanager
from typing import Generator
import time

@contextmanager
def timed_block(label: str) -> Generator[None, None, None]:
    """计时一个代码块"""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"⏱ [{label}] 耗时: {elapsed:.4f}s")

# 使用
with timed_block("数据处理"):
    time.sleep(0.5)
    # ... 数据处理代码 ...

9. 装饰器 vs 上下文管理器：如何选择

维度	装饰器	上下文管理器
关注点	修改/增强函数	管理资源生命周期
代码位置	定义函数时	使用资源时
适用范围	整个函数调用	特定的代码块
典型场景	日志、缓存、重试、权限	文件、锁、事务、临时配置
可组合性	可叠加多个装饰器	可嵌套多个 with

场景决策

需要在函数调用前后添加行为？
├── 与特定资源相关（文件/锁/连接）
│   └── → 上下文管理器
└── 是通用横切关注点（日志/计时/缓存）
    └── → 装饰器

需要管理一个代码块内的资源？
└── → 上下文管理器

需要给一个函数附加元数据/行为？
└── → 装饰器

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页