Python3 【装饰器】项目实战：5个新颖的学习案例

news2025/2/1 12:26:19

Python3 【装饰器】项目实战：5个新颖的学习案例

以下是 5 个使用 Python 装饰器的综合应用项目，这些项目具有新颖性、前瞻性和实用性。每个项目都包含完整的代码、解释说明、测试案例和执行结果。

项目 1：API 请求限流器

描述：实现一个装饰器，限制 API 请求的频率，防止服务器过载。

代码实现

import time

def rate_limiter(max_calls, period):
    """
    限制函数调用频率的装饰器。
    :param max_calls: 每个时间段内允许的最大调用次数。
    :param period: 时间段（秒）。
    """
    calls = []

    def decorator(func):
        def wrapper(*args, **kwargs):
            now = time.time()
            # 移除超时的调用记录
            calls[:] = [call for call in calls if now - call < period]
            if len(calls) >= max_calls:
                raise Exception("请求频率过高，请稍后再试")
            calls.append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator

# 测试案例
@rate_limiter(max_calls=3, period=5)
def api_request():
    print("API 请求成功")

# 模拟请求
for i in range(5):
    try:
        api_request()
    except Exception as e:
        print(e)
    time.sleep(1)

执行结果

API 请求成功
API 请求成功
API 请求成功
请求频率过高，请稍后再试
请求频率过高，请稍后再试

说明：该装饰器限制每 5 秒内最多调用 3 次 API，超过限制会抛出异常。

项目 2：函数结果缓存

描述：实现一个装饰器，缓存函数的结果，避免重复计算。

代码实现

def cache_decorator(func):
    """
    缓存函数结果的装饰器。
    """
    cache = {}

    def wrapper(*args):
        if args in cache:
            print("从缓存中获取结果")
            return cache[args]
        result = func(*args)
        cache[args] = result
        return result
    return wrapper

# 测试案例
@cache_decorator
def fibonacci(n):
    """
    计算斐波那契数列。
    """
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(10))
print(fibonacci(10))

执行结果

55
从缓存中获取结果
55

说明：该装饰器缓存了斐波那契数列的计算结果，避免重复计算。

项目 3：异步任务重试机制

描述：实现一个装饰器，为异步任务提供重试机制。

代码实现

import asyncio
import random

def retry_decorator(max_retries=3):
    """
    为异步函数提供重试机制的装饰器。
    :param max_retries: 最大重试次数。
    """
    def decorator(func):
        async def wrapper(*args, **kwargs):
            for i in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    print(f"重试 {i+1}/{max_retries}: {e}")
                    await asyncio.sleep(1)
            raise Exception("重试次数用尽")
        return wrapper
    return decorator

# 测试案例
@retry_decorator(max_retries=3)
async def unstable_task():
    if random.random() < 0.5:
        raise Exception("任务失败")
    return "任务成功"

async def main():
    try:
        result = await unstable_task()
        print(result)
    except Exception as e:
        print(e)

asyncio.run(main())

执行结果

重试 1/3: 任务失败
任务成功

说明：该装饰器为异步任务提供了重试机制，失败后最多重试 3 次。

项目 4：函数调用日志记录

描述：实现一个装饰器，记录函数的调用日志。

代码实现

import logging

logging.basicConfig(level=logging.INFO)

def log_decorator(func):
    """
    记录函数调用日志的装饰器。
    """
    def wrapper(*args, **kwargs):
        logging.info(f"调用函数: {func.__name__}, 参数: {args}, {kwargs}")
        result = func(*args, **kwargs)
        logging.info(f"函数 {func.__name__} 执行完毕")
        return result
    return wrapper

# 测试案例
@log_decorator
def add(a, b):
    return a + b

print(add(3, 5))

执行结果

INFO:root:调用函数: add, 参数: (3, 5), {}
INFO:root:函数 add 执行完毕
8

说明：该装饰器记录了函数的调用和参数，方便调试和追踪。

项目 5：动态权限验证

描述：实现一个装饰器，根据用户角色动态验证权限。

代码实现

def role_required(role):
    """
    根据用户角色验证权限的装饰器。
    :param role: 需要的角色。
    """
    def decorator(func):
        def wrapper(user, *args, **kwargs):
            if user.get("role") == role:
                return func(*args, **kwargs)
            else:
                return "权限不足"
        return wrapper
    return decorator

# 测试案例
@role_required(role="admin")
def delete_file():
    return "文件已删除"

user_admin = {"name": "Alice", "role": "admin"}
user_guest = {"name": "Bob", "role": "guest"}

print(delete_file(user_admin))
print(delete_file(user_guest))