wordflow
返回博客

23. 装饰器与偏函数：代码复用的利器

23. 装饰器与偏函数：代码复用的利器

第二十三篇：装饰器与偏函数——代码复用的利器

前言

**装饰器（Decorator）和偏函数（partial）**是 Python 中实现代码复用和增强功能的两大核心机制。本文深入讲解装饰器的原理、常见模式以及偏函数的应用场景。

装饰器基础

什么是装饰器？

装饰器 = 接受函数作为参数并返回新函数的函数。核心作用是在不修改原函数源码的情况下增强其行为。

def my_decorator(func):
    """装饰器：在调用前后添加日志"""
    def wrapper(*args, **kwargs):
        print(f"[前置] 调用 {func.__name__}")
        result = func(*args, **kwargs)
        print(f"[后置] 得到结果：{result}")
        return result
    return wrapper

# 使用 @ 语法糖
@my_decorator
def add(a, b):
    return a + b

print(add(3, 5))
# [前置] 调用 add
# [后置] 得到结果：8
# 输出：8

装饰器的内部原理

# 以下两种方式等价：

# 方式一：使用 @ 语法糖
@my_decorator
def greet(name):
    return f"Hello, {name}!"

# 方式二：手动赋值
def greet(name):
    return f"Hello, {name}!"
greet = my_decorator(greet)

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

functools.wraps — 保留元信息

import functools

def logging_decorator(func):
    @functools.wraps(func)  # ← 关键：保留 __name__, __doc__ 等元数据
    def wrapper(*args, **kwargs):
        """Wrapper 的 docstring"""
        print(f"Calling {func.__name__}...")
        return func(*args, **kwargs)
    
    wrapper.__original_doc__ = func.__doc__  # 额外保留原文档
    return wrapper

@logging_decorator
def calculate(x, y):
    """计算两个数之和"""
    return x + y

print(calculate.__name__)   # calculate — 不是 wrapper！
print(calculate.__doc__)    # 计算两个数之和

实战装饰器模式

计时器装饰器

import time
import functools

def timer(func):
    """测量函数执行时间"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()
        elapsed = end - start
        
        # 根据耗时选择不同的提示语
        if elapsed < 0.001:
            unit, display = "μs", int(elapsed * 1_000_000)
        elif elapsed < 1:
            unit, display = "ms", int(elapsed * 1_000)
        else:
            unit, display = "s", round(elapsed, 2)
        
        print(f"{func.__name__:20s} [{unit:2s}] {display:>8}")
        return result
    
    wrapper.elapsed_times = []  # 记录历史耗时
    return wrapper

@timer
def sort_list(n):
    import random
    data = [random.randint(0, 1000) for _ in range(n)]
    return sorted(data)

@timer
def find_max(n):
    import random
    data = [random.randint(0, 1000) for _ in range(n)]
    return max(data)

sort_list(10000)
find_max(10000)

重试装饰器

import functools
import time

def retry(max_retries=3, delay=1, backoff=2):
    """自动重试装饰器，支持指数退避"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            current_delay = delay
            
            while True:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    if attempts >= max_retries:
                        raise
                    
                    print(f"失败 ({attempts}/{max_retries}): {e}")
                    time.sleep(current_delay)
                    current_delay *= backoff  # 指数退避
        
        wrapper.retries = max_retries
        wrapper.current_delay = current_delay
        return wrapper
    
    return decorator

class NetworkService:
    @retry(max_retries=3, delay=0.5, backoff=2)
    def fetch_data(self):
        import random
        if random.random() < 0.7:  # 70% 概率失败
            raise ConnectionError("网络超时")
        return "成功获取数据"

service = NetworkService()
try:
    print(service.fetch_data())
except ConnectionError as e:
    print(f"最终失败：{e}")

缓存/记忆化装饰器

from functools import lru_cache

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

for i in range(30):
    print(f"F({i}) = {fibonacci(i)}", end=" ")
print(f"\n缓存条目数：{fibonacci.cache_info()}")  # 仅 30 个唯一输入

自定义记忆化装饰器：

def memoize(func):
    """通用记忆化装饰器"""
    cache = {}
    
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    
    wrapper.cache_info = lambda: len(cache)
    wrapper.clear_cache = lambda: cache.clear()
    return wrapper

@memoize
def expensive_function(x):
    total = sum(range(x * 1000000))
    return total

print(expensive_function(5))   # 第一次：慢
print(expensive_function(5))   # 第二次：瞬间！（命中缓存）
expensive_function.clear_cache()  # 清除缓存

权限验证装饰器

import functools

def require_role(*roles):
    """角色权限验证装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(user, *args, **kwargs):
            if user.get("role") not in roles:
                raise PermissionError(
                    f"角色 '{user.get('role')}' 无权访问 {func.__name__}"
                )
            return func(user, *args, **kwargs)
        return wrapper
    return decorator

def check_permissions(required_level):
    """权限级别装饰器工厂"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(user, *args, **kwargs):
            user_level = user.get("level", 0)
            if user_level < required_level:
                raise PermissionError(
                    f"需要 level≥{required_level}，当前 {user_level}"
                )
            return func(user, *args, **kwargs)
        return wrapper
    return decorator

class AdminUser:
    role = "admin"
    level = 5

class RegularUser:
    role = "user"
    level = 1

@require_role("admin", "moderator")
@check_permissions(3)
def admin_panel(user, panel_name):
    return f"管理员 {user['role']} 正在查看 {panel_name}"

try:
    print(admin_panel(AdminUser(), "系统设置"))
    # 管理员 admin 正在查看 系统设置
except PermissionError as e:
    print(f"权限错误：{e}")

try:
    print(admin_panel(RegularUser(), "系统设置"))
except PermissionError as e:
    print(f"权限错误：{e}")

带参数的装饰器

import functools

def repeat(times=1):
    """重复执行指定次数"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for i in range(times):
                result = func(*args, **kwargs)
                results.append(result)
                print(f"执行 {i+1}/{times}")
            return results if times > 1 else results[0]
        return wrapper
    return decorator

@repeat(times=3)
def greet(name):
    print(f"你好，{name}！")
    return name

greet("小明")
# 输出：
# 执行 1/3
# 你好，小明！
# 执行 2/3
# 你好，小明！
# 执行 3/3
# 你好，小明！

偏函数 partial

functools.partial 简介

偏函数 = 部分应用函数参数，返回一个新的函数对象。

from functools import partial

def power(base, exponent):
    return base ** exponent

# 固定 exponent 参数，创建专用函数
square = partial(power, exponent=2)      # x²
cube = partial(power, exponent=3)        # x³
sqrt = partial(power, exponent=0.5)      # √x

print(square(5))   # 25
print(cube(3))     # 27
print(sqrt(16))    # 4.0

# 底层等价于：lambda x: power(x, exponent=2)

偏函数实战场景

# 场景一：减少回调函数的参数数量
import time

def send_notification(to, message, method="email", timestamp=None):
    if timestamp is None:
        timestamp = time.time()
    print(f"[{timestamp}] 发送 {method} 给 {to}: {message}")

# 为特定用户预配置通知方式
notify_alice_email = partial(send_notification, "alice@example.com", "密码重置链接", "email")
notify_bob_sms = partial(send_notification, "+86138xxxx", "验证码", "sms")

notify_alice_email()
notify_bob_sms()
# [1725684000.0] 发送 email 给 alice@example.com: 密码重置链接
# [1725684000.0] 发送 sms 给 +86138xxxx: 验证码

# 场景二：简化 map/filter/reduce 调用
nums = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

# 传统方式
doubled = list(map(lambda x: x * 2, nums))

# 使用 partial（更简洁）
double = partial(lambda x, y: x * y, y=2)
doubled_v2 = list(map(double, nums))
print(doubled_v2)  # [2, 4, 6, 8, 10, 12, 14, 16, 18, 20]

装饰器组合

多个装饰器叠加

import functools

def logger(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print(f"[{func.__name__}] 开始")
        result = func(*args, **kwargs)
        print(f"[{func.__name__}] 结束")
        return result
    return wrapper

def counter(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        wrapper.call_count += 1
        print(f"调用 #{wrapper.call_count}")
        return func(*args, **kwargs)
    wrapper.call_count = 0
    return wrapper

@counter
@logger
def complex_operation():
    print("正在进行计算...")
    return sum(range(1_000_000))

complex_operation()
# 输出：
# 调用 #1
# [complex_operation] 开始
# 正在进行计算...
# [complex_operation] 结束

装饰器顺序的重要性

def uppercase(func):
    def wrapper(s):
        return func(s.upper())
    return wrapper

def repeat_n(n):
    def decorator(func):
        def wrapper(s):
            result = func(s)
            return result * n
        return wrapper
    return decorator

# 顺序不同，行为完全不同
@repeat_n(2)
@uppercase
def shout(s):
    return s + "!"

print(shout("hello"))   # HELLO!HELLO!

# 如果反过来：
@uppercase
@repeat_n(2)
def whisper(s):
    return s + "?"

print(whisper("hello"))  # ???

常见问题与注意事项

陷阱一：装饰器改变函数签名

def trace(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper  # ❌ 丢失了原始签名

@trace
def add(a, b):
    pass

help(add)  # wrapper 的签名而非 add 的

修复：使用 @functools.wraps(func)

陷阱二：异步函数不能用普通装饰器

import asyncio

async def slow_request(url):
    await asyncio.sleep(1)
    return f"Response from {url}"

# ❌ 普通装饰器无法正确包裹异步函数
# @logging_decorator
# async def fetch_page():
#     pass

# ✅ 使用 aiocontextlib 或编写异步装饰器

陷阱三：闭包变量捕获延迟

# ❌ 所有装饰器共享同一个 loop_counter
loop_counters = []
for i in range(5):
    def make_counter():
        count = [0]  # 被所有闭包共享
        def increment():
            count[0] += 1
            return count[0]
        return increment
    loop_counters.append(make_counter())

# ✅ 每个装饰器应独立初始化
counters = [lambda x=i: x for i in range(5)]

总结

本文涵盖了装饰器与偏函数的核心知识：

  1. 装饰器基础 — 接收函数、返回新函数、@ 语法糖
  2. wraps 装饰器 — 保留原始函数元数据
  3. 计时器 — 测量执行时间的通用装饰器
  4. 重试装饰器 — 指数退避的容错机制
  5. 缓存装饰器 — lru_cache/自定义记忆化
  6. 权限验证 — 角色/级别控制装饰器
  7. 偏函数 — partial 减少参数传递
  8. 装饰器组合 — @decorator1 then @decorator2
  9. 最佳实践 — wraps 不可少、异步需特殊处理
  10. 性能对比 — 装饰器有轻微 overhead，但提升开发效率显著

下一篇文章将深入讲解模块系统：导入、作用域与包结构。

本文是 Python 教程系列的第二十三篇。