Python 装饰器10种高级用法，用完后悔没早知道

用户11081884

发布于 2026-07-20 20:16:33

100

文章被收录于专栏：用户11081884的专栏用户11081884的专栏

Python装饰器是函数式编程的重要特性，它允许在不修改原函数代码的情况下，动态地扩展函数功能。以下是10种高级用法，每种都包含说明、使用场景和代码示例。

1. 带参数的装饰器

说明：装饰器本身可以接受参数，通过外层函数传递参数实现更灵活的装饰逻辑。

使用场景：需要根据配置动态调整装饰行为的场景，如不同环境的日志级别、缓存时间等。

defretry(max_attempts=3):
    defdecorator(func):
        defwrapper(*args, **kwargs):
            forattemptinrange(max_attempts):
                try:
                    returnfunc(*args, **kwargs)
                exceptExceptionase:
                    ifattempt == max_attempts-1:
                        raisee
                    print(f"Attempt {attempt + 1} failed, retrying...")
            returnNone
        returnwrapper
    returndecorator

@retry(max_attempts=5)
deffetch_data(url):
    # 模拟网络请求
    importrandom
    ifrandom.random() <0.7:
        raiseConnectionError("Connection failed")
    return"Data fetched successfully"

print(fetch_data("https://api.example.com"))

2. 类装饰器

说明：使用类定义装饰器逻辑，通过__init__和__call__方法实现。

使用场景：需要维护状态的装饰器，如计数器、缓存管理等。

classCountCalls:
    def__init__(self, func):
        self.func = func
        self.call_count = 0
    
    def__call__(self, *args, **kwargs):
        self.call_count += 1
        print(f"{self.func.__name__} has been called {self.call_count} times")
        returnself.func(*args, **kwargs)

@CountCalls
defprocess_data(data):
    returndata.upper()

process_data("hello")
process_data("world")

3. 装饰器链

说明：多个装饰器按从外到内顺序应用于同一函数。

使用场景：需要组合多个功能的场景，如日志+缓存+权限验证。

deflog_execution(func):
    defwrapper(*args, **kwargs):
        print(f"Executing {func.__name__} with args: {args}, kwargs: {kwargs}")
        result = func(*args, **kwargs)
        print(f"{func.__name__} returned: {result}")
        returnresult
    returnwrapper

defvalidate_input(func):
    defwrapper(*args, **kwargs):
        forarginargs:
            ifnotisinstance(arg, (int, float)):
                raiseValueError("Arguments must be numbers")
        returnfunc(*args, **kwargs)
    returnwrapper

@log_execution
@validate_input
defadd_numbers(a, b):
    returna+b

print(add_numbers(5, 3))

4. 装饰器工厂（装饰器中的装饰器）

说明：装饰器内部使用其他装饰器，实现功能组合。

使用场景：创建可复用的装饰器组合。

deftimer_decorator(func):
    importtime
    defwrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} took {end - start:.4f} seconds")
        returnresult
    returnwrapper

deflog_decorator(func):
    defwrapper(*args, **kwargs):
        print(f"Calling {func.__name__}")
        returnfunc(*args, **kwargs)
    returnwrapper

defcombined_decorator(func):
    # 组合timer和log装饰器
    returntimer_decorator(log_decorator(func))

@combined_decorator
defheavy_computation(n):
    importtime
    time.sleep(0.5)
    returnsum(range(n))

print(heavy_computation(1000))

5. 异步函数装饰器

说明：专门用于装饰异步函数的装饰器。

使用场景：异步编程中的性能监控、错误重试等。

importasyncio

defasync_timer(func):
    asyncdefwrapper(*args, **kwargs):
        importtime
        start = time.time()
        result = awaitfunc(*args, **kwargs)
        end = time.time()
        print(f"Async {func.__name__} took {end - start:.4f} seconds")
        returnresult
    returnwrapper

@async_timer
asyncdeffetch_urls(urls):
    importaiohttp
    asyncwithaiohttp.ClientSession() assession:
        tasks = [session.get(url) forurlinurls]
        responses = awaitasyncio.gather(*tasks)
        return [awaitr.text() forrinresponses]

# 使用示例
# asyncio.run(fetch_urls(["http://example.com"]))

6. 属性装饰器

说明：在类方法上使用装饰器访问类属性。

使用场景：类方法的状态管理、访问控制等。

classRateLimiter:
    def__init__(self, max_calls=3):
        self.max_calls = max_calls
        self.call_count = 0
    
    deflimit_calls(self, func):
        defwrapper(*args, **kwargs):
            ifself.call_count>= self.max_calls:
                raiseException(f"Rate limit exceeded. Max calls: {self.max_calls}")
            self.call_count += 1
            returnfunc(*args, **kwargs)
        returnwrapper

limiter = RateLimiter(max_calls=2)

classAPI:
    @limiter.limit_calls
    defcall_endpoint(self, endpoint):
        returnf"Calling {endpoint}"

api = API()
print(api.call_endpoint("/users"))
print(api.call_endpoint("/posts"))
# 第三次调用会触发异常

7. 单例模式装饰器

说明：确保类只有一个实例的装饰器。

使用场景：数据库连接、配置管理、日志记录器等需要单例的场景。

defsingleton(cls):
    instances = {}
    defwrapper(*args, **kwargs):
        ifclsnotininstances:
            instances[cls] = cls(*args, **kwargs)
        returninstances[cls]
    returnwrapper

@singleton
classDatabaseConnection:
    def__init__(self):
        print("Creating new database connection...")
        self.connection_id = id(self)
    
    defquery(self, sql):
        returnf"Executing: {sql}"

# 测试
db1 = DatabaseConnection()
db2 = DatabaseConnection()
print(f"db1 id: {db1.connection_id}")
print(f"db2 id: {db2.connection_id}")
print(f"Same instance: {db1 is db2}")

8. 类型检查装饰器

说明：在运行时检查函数参数和返回值的类型。

使用场景：调试、API接口验证、类型安全要求高的场景。

deftype_check(*arg_types, return_type=None):
    defdecorator(func):
        defwrapper(*args, **kwargs):
            # 检查参数类型
            fori, (arg, expected_type) inenumerate(zip(args, arg_types)):
                ifnotisinstance(arg, expected_type):
                    raiseTypeError(f"Argument {i} must be {expected_type}, got {type(arg)}")
            
            # 执行函数
            result = func(*args, **kwargs)
            
            # 检查返回值类型
            ifreturn_typeandnotisinstance(result, return_type):
                raiseTypeError(f"Return value must be {return_type}, got {type(result)}")
            
            returnresult
        returnwrapper
    returndecorator

@type_check(int, int, return_type=int)
defmultiply(a, b):
    returna*b

print(multiply(5, 3))  # 正常
# print(multiply(5, "3"))  # 会抛出TypeError

9. 缓存装饰器

说明：缓存函数结果，避免重复计算。

使用场景：计算成本高、结果不变的函数，如API调用、复杂计算等。

fromfunctoolsimportlru_cache
importtime

# 使用functools.lru_cache
@lru_cache(maxsize=128)
deffibonacci(n):
    ifn<2:
        returnn
    returnfibonacci(n-1) +fibonacci(n-2)

# 自定义缓存装饰器
defcache_results(func):
    cache = {}
    defwrapper(*args, **kwargs):
        key = (args, tuple(kwargs.items()))
        ifkeynotincache:
            cache[key] = func(*args, **kwargs)
        returncache[key]
    returnwrapper

@cache_results
defexpensive_computation(x):
    time.sleep(1)  # 模拟耗时计算
    returnx*x

# 第一次调用耗时
start = time.time()
print(expensive_computation(5))
print(f"First call: {time.time() - start:.2f}s")

# 第二次调用从缓存读取
start = time.time()
print(expensive_computation(5))
print(f"Second call: {time.time() - start:.2f}s")

10. 权限验证装饰器

说明：验证用户权限后再执行函数。

使用场景：Web应用、API接口的权限控制。

defrequire_role(*allowed_roles):
    defdecorator(func):
        defwrapper(user, *args, **kwargs):
            ifuser.get('role') notinallowed_roles:
                raisePermissionError(
                    f"User {user.get('name')} with role {user.get('role')} "
                    f"is not allowed to call {func.__name__}. "
                    f"Allowed roles: {allowed_roles}"
                )
            returnfunc(user, *args, **kwargs)
        returnwrapper
    returndecorator

classUserService:
    @require_role('admin', 'manager')
    defdelete_user(self, user, user_id):
        returnf"User {user_id} deleted by {user['name']}"
    
    @require_role('admin', 'manager', 'user')
    defview_profile(self, user, user_id):
        returnf"Viewing profile of user {user_id}"

service = UserService()
admin_user = {'name': 'Alice', 'role': 'admin'}
regular_user = {'name': 'Bob', 'role': 'user'}

print(service.delete_user(admin_user, 123))  # 正常
print(service.view_profile(regular_user, 123))  # 正常
# print(service.delete_user(regular_user, 123))  # 会抛出PermissionError

实战应用：完整的API装饰器组合

importtime
importfunctools

defapi_decorator(func):
    """API接口的完整装饰器组合"""
    @functools.wraps(func)
    defwrapper(*args, **kwargs):
        # 1. 日志记录
        print(f"[API] {time.strftime('%Y-%m-%d %H:%M:%S')} - Calling {func.__name__}")
        
        # 2. 参数验证
        ifnotargsandnotkwargs:
            print("[API] Warning: No arguments provided")
        
        # 3. 性能监控
        start_time = time.time()
        
        try:
            # 执行原函数
            result = func(*args, **kwargs)
            
            # 4. 响应时间记录
            elapsed = time.time() -start_time
            print(f"[API] {func.__name__} completed in {elapsed:.4f}s")
            
            # 5. 结果验证
            ifresultisNone:
                print("[API] Warning: Function returned None")
            
            returnresult
            
        exceptExceptionase:
            # 6. 错误处理
            print(f"[API] Error in {func.__name__}: {str(e)}")
            raise
    
    returnwrapper

@api_decorator
defget_user_data(user_id, include_profile=False):
    """获取用户数据"""
    ifnotisinstance(user_id, int):
        raiseValueError("user_id must be an integer")
    
    # 模拟API调用
    time.sleep(0.1)
    
    data = {
        'id': user_id,
        'name': f'User{user_id}',
        'email': f'user{user_id}@example.com'
    }
    
    ifinclude_profile:
        data['profile'] = {'age': 25, 'city': 'Beijing'}
    
    returndata

# 使用示例
try:
    user_data = get_user_data(123, include_profile=True)
    print(f"User data: {user_data}")
exceptExceptionase:
    print(f"Failed to get user data: {e}")

Python装饰器的10种高级用法展示了其强大的灵活性和实用性：