Python 状态保持装饰器

状态保持装饰器在多次调用间保存信息，实现计数、缓存、限频等功能。

函数计数装饰器

def count_calls(func):
    count = 0  # 闭包保存状态

    def wrapper(*args, **kwargs):
        nonlocal count
        count += 1
        print(f"{func.__name__} 被调用 {count} 次")
        return func(*args, **kwargs)

    return wrapper

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

greet('Alice')  # greet 被调用 1 次
greet('Bob')    # greet 被调用 2 次
greet('Charlie')# greet 被调用 3 次

类计数装饰器

class CountCalls:
    def __init__(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 greet(name):
    return f"Hello, {name}"

greet('Alice')
greet('Bob')

# 访问状态
print(greet.count)  # 2

缓存装饰器

def cache(func):
    cached = {}  # 闭包保存缓存

    def wrapper(*args):
        if args in cached:
            print(f"缓存命中: {args}")
            return cached[args]
        result = func(*args)
        cached[args] = result
        return result

    return wrapper

@cache
def expensive_calc(n):
    print(f"计算 {n}")
    return n ** 2

expensive_calc(5)  # 计算 5
expensive_calc(5)  # 缓存命中: (5,)
expensive_calc(10) # 计算 10
expensive_calc(5)  # 缓存命中: (5,)

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)

# 查看缓存状态
print(fibonacci.cache_info())
# CacheInfo(hits=0, misses=101, maxsize=128, currsize=101)

fibonacci(100)
print(fibonacci.cache_info())

# 清除缓存
fibonacci.cache_clear()

调用频率限制

import time

def rate_limit(max_calls, period):
    calls = []  # 记录调用时间

    def decorator(func):
        def wrapper(*args, **kwargs):
            now = time.time()

            # 清除过期记录
            calls[:] = [t for t in calls if now - t < period]

            if len(calls) >= max_calls:
                raise RuntimeError(f"超过调用限制: {max_calls}/{period}s")

            calls.append(now)
            return func(*args, **kwargs)

        return wrapper
    return decorator

@rate_limit(3, 1.0)  # 每秒最多3次
def limited_func():
    return "执行成功"

limited_func()  # 成功
limited_func()  # 成功
limited_func()  # 成功
# limited_func()  # RuntimeError

类实现限频

import time

class RateLimit:
    def __init__(self, max_calls, period):
        self.max_calls = max_calls
        self.period = period
        self.calls = []

    def __call__(self, func):
        def wrapper(*args, **kwargs):
            now = time.time()
            self.calls[:] = [t for t in self.calls if now - t < self.period]

            if len(self.calls) >= self.max_calls:
                raise RuntimeError("超过限制")

            self.calls.append(now)
            return func(*args, **kwargs)

        wrapper.calls = self.calls  # 可访问状态
        return wrapper

@RateLimit(5, 10.0)
def api_call():
    return "API调用成功"

计时统计装饰器

import time

class TimerStats:
    def __init__(self, func):
        self.func = func
        self.total_time = 0
        self.call_count = 0

    def __call__(self, *args, **kwargs):
        start = time.time()
        result = self.func(*args, **kwargs)
        elapsed = time.time() - start

        self.total_time += elapsed
        self.call_count += 1

        return result

    def stats(self):
        avg = self.total_time / self.call_count if self.call_count else 0
        return {
            'calls': self.call_count,
            'total': self.total_time,
            'average': avg
        }

@TimerStats
def process_data(n):
    return sum(range(n))

process_data(10000)
process_data(10000)

print(process_data.stats())
# {'calls': 2, 'total': 0.001, 'average': 0.0005}

重试装饰器

def retry(max_attempts, delay=1):
    attempts = 0

    def decorator(func):
        def wrapper(*args, **kwargs):
            nonlocal attempts
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    if attempts >= max_attempts:
                        raise
                    time.sleep(delay)
        return wrapper
    return decorator

# 类实现更清晰
class Retry:
    def __init__(self, max_attempts, delay=1):
        self.max_attempts = max_attempts
        self.delay = delay
        self.attempts = 0

    def __call__(self, func):
        def wrapper(*args, **kwargs):
            self.attempts = 0
            while self.attempts < self.max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception:
                    self.attempts += 1
                    if self.attempts >= self.max_attempts:
                        raise
                    time.sleep(self.delay)
        wrapper.attempts = self.attempts
        return wrapper

状态保持方式对比

状态访问

# 闭包方式：状态封装在内部
@count_calls
def func():
    pass
# 无法直接访问 count

# 类方式：状态可访问
@CountCalls
def func():
    pass
print(func.count)  # 可访问状态

要点总结

• 闭包变量保存装饰器状态
• nonlocal 声明修改闭包变量
• 类装饰器状态可通过属性访问
• __call__ 使类实例可调用
• 计数器、缓存、限频需要状态保持
• lru_cache 内置缓存状态管理
• 类装饰器比闭包更易访问状态
• 状态保持装饰器实现跨调用功能累计