如何在Python中将类属性用作装饰器？以限流器为例

我需要实现一个Python的RateLimiter类，作为其他类的实例属性，并用装饰器形式给需要限流的函数加限制。现有RateLimiter实现如下：

import time
from functools import wraps
from threading import Lock

class RateLimiter:
    def __init__(self, qps: float):
        self.qps = qps
        self.min_interval = 1.0 / qps
        self.last_call = 0.0
        self.lock = Lock()

    def limit(self):
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                with self.lock:
                    now = time.monotonic()
                    elapsed = now - self.last_call
                    if elapsed < self.min_interval:
                        time.sleep(self.min_interval - elapsed)
                    self.last_call = time.monotonic()
                return func(*args, **kwargs)
            return wrapper
        return decorator

我原本期望在SomeWebClient类里这样用：

class SomeWebClient:
    def __init__(self, qps: float = 10.0):
        self.rate_limiter = RateLimiter(qps=qps)

    @self.rate_limiter.limit()
    def some_api_call(self):
        # 调用受限流的API
        pass

    @self.rate_limiter.limit()
    def another_api_call(self):
        # 调用受限流的API
        pass

但这段代码无法运行，因为类定义阶段self还未定义。目前只能用这种不够简洁的方式实现：

class SomeWebClient:
    def __init__(self, qps: float = 10.0):
        self.rate_limiter = RateLimiter(qps=qps)
        self.some_api_call = self.rate_limiter.limit()(self.some_api_call)
        self.another_api_call = self.rate_limiter.limit()(self.another_api_call)

    def some_api_call(self):
        # 调用受限流的API
        pass

    def another_api_call(self):
        # 调用受限流的API
        pass

需求：每个SomeWebClient实例拥有独立限流器，可设置不同QPS，且能清晰标记需要限流的方法，希望能用@语法实现，或者其他简洁的非装饰器方案。

解决方案 #

方案1：标记装饰器+实例初始化自动绑定 #

先定义一个空的标记装饰器标记需要限流的方法，再在实例初始化时自动为这些方法绑定限流器：

import time
from functools import wraps
from threading import Lock

class RateLimiter:
    def __init__(self, qps: float):
        self.qps = qps
        self.min_interval = 1.0 / qps
        self.last_call = 0.0
        self.lock = Lock()

    def limit(self):
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                with self.lock:
                    now = time.monotonic()
                    elapsed = now - self.last_call
                    if elapsed < self.min_interval:
                        time.sleep(self.min_interval - elapsed)
                    self.last_call = time.monotonic()
                return func(*args, **kwargs)
            return wrapper
        return decorator

# 标记需要限流的方法
def needs_rate_limit(func):
    func._needs_rate_limit = True
    return func

class SomeWebClient:
    def __init__(self, qps: float = 10.0):
        self.rate_limiter = RateLimiter(qps=qps)
        # 遍历实例方法，给标记过的方法应用限流
        for name, method in vars(self).items():
            if callable(method) and hasattr(method, "_needs_rate_limit"):
                setattr(self, name, self.rate_limiter.limit()(method))

    @needs_rate_limit
    def some_api_call(self):
        print("Executing some_api_call")

    @needs_rate_limit
    def another_api_call(self):
        print("Executing another_api_call")

这种方式用@needs_rate_limit清晰标记目标方法，初始化逻辑自动完成限流绑定，保持代码简洁。

方案2：描述符实现实例级装饰器 #

利用Python描述符协议，让装饰器在调用时自动获取实例的限流器：

import time
from functools import wraps
from threading import Lock

class RateLimiter:
    def __init__(self, qps: float):
        self.qps = qps
        self.min_interval = 1.0 / qps
        self.last_call = 0.0
        self.lock = Lock()

    def limit(self, func):
        @wraps(func)
        def wrapper(self_instance, *args, **kwargs):
            # self_instance为SomeWebClient的实例
            with self.lock:
                now = time.monotonic()
                elapsed = now - self.last_call
                if elapsed < self.min_interval:
                    time.sleep(self.min_interval - elapsed)
                self.last_call = time.monotonic()
            return func(self_instance, *args, **kwargs)
        return wrapper

# 描述符类，用于绑定实例限流器
class InstanceRateLimit:
    def __init__(self, limiter_attr_name):
        self.limiter_attr_name = limiter_attr_name
        self.func = None

    def __call__(self, func):
        self.func = func
        return self

    def __get__(self, instance, owner):
        if instance is None:
            return self
        # 获取实例的限流器并应用
        limiter = getattr(instance, self.limiter_attr_name)
        return limiter.limit(self.func)

class SomeWebClient:
    def __init__(self, qps: float = 10.0):
        self.rate_limiter = RateLimiter(qps=qps)

    @InstanceRateLimit("rate_limiter")
    def some_api_call(self):
        print("Executing some_api_call")

    @InstanceRateLimit("rate_limiter")
    def another_api_call(self):
        print("Executing another_api_call")

直接用@InstanceRateLimit("rate_limiter")装饰方法，指定实例中限流器的属性名，调用时自动完成限流逻辑，完全符合@语法的使用习惯。

方案3：类装饰器批量处理（多类复用场景） #

如果多个类需要限流功能，可用类装饰器批量处理标记方法：

import time
from functools import wraps
from threading import Lock

class RateLimiter:
    def __init__(self, qps: float):
        self.qps = qps
        self.min_interval = 1.0 / qps
        self.last_call = 0.0
        self.lock = Lock()

    def limit(self):
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                with self.lock:
                    now = time.monotonic()
                    elapsed = now - self.last_call
                    if elapsed < self.min_interval:
                        time.sleep(self.min_interval - elapsed)
                    self.last_call = time.monotonic()
                return func(*args, **kwargs)
            return wrapper
        return decorator

# 类装饰器，自动处理限流绑定
def rate_limited_class(limiter_attr="rate_limiter"):
    def decorator(cls):
        original_init = cls.__init__
        def new_init(self, *args, **kwargs):
            original_init(self, *args, **kwargs)
            limiter = getattr(self, limiter_attr)
            for name, method in vars(cls).items():
                if callable(method) and hasattr(method, "_needs_rate_limit"):
                    setattr(self, name, limiter.limit()(method))
        cls.__init__ = new_init
        return cls
    return decorator

def needs_rate_limit(func):
    func._needs_rate_limit = True
    return func

@rate_limited_class()
class SomeWebClient:
    def __init__(self, qps: float = 10.0):
        self.rate_limiter = RateLimiter(qps=qps)

    @needs_rate_limit
    def some_api_call(self):
        print("Executing some_api_call")

    @needs_rate_limit
    def another_api_call(self):
        print("Executing another_api_call")

该方案适合多类复用场景，类装饰器自动完成初始化时的限流绑定，方法只需标记@needs_rate_limit即可。

内容的提问来源于stack exchange，提问作者ROODAY