20个简单实用的Python装饰器

上篇介绍了Python装饰器及相关学习资源:Python高阶函数与装饰器,这次整了网上一些简单实用的Python装饰器示例。

1 @timer:测量执行时间

优化代码性能是非常重要的。@timer装饰器可以帮助我们跟踪特定函数的执行时间。通过用这个装饰器包装函数,可以快速识别瓶颈并优化代码的关键部分。下面是它的工作原理:

 import time
 
 def timer(func):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__} took {end_time - start_time:.2f} seconds to execute.")
        return result
    return wrapper
 
 @timer
 def my_data_processing_function():
    # Your data processing code here

将@timer与其他装饰器结合使用,可以全面地分析代码的性能。

2 @memoize:缓存结果

在数据科学中,我们经常使用计算成本很高的函数。@memoize装饰器可以帮助缓存函数结果,避免了相同输入的冗余计算,显著加快工作流程:
 def memoize(func):
    cache = {}
 
 def wrapper(*args):
        if args in cache:
            return cache[args]
        result = func(*args)
        cache[args] = result
        return result
    return wrapper
 
 @memoize
 def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)
在递归函数中也可以使用@memoize来优化重复计算。

3 @validate_input:数据验证

数据完整性至关重要,@validate_input装饰器可以验证函数参数,确保它们在继续计算之前符合特定的标准:

 def validate_input(func):
    def wrapper(*args, **kwargs):
        # Your data validation logic here
        if valid_data:
            return func(*args, **kwargs)
        else:
            raise ValueError("Invalid data. Please check your inputs.")
    return wrapper
 
 @validate_input
 def analyze_data(data):
    # Your data analysis code here

可以方便的使用@validate_input在数据科学项目中一致地实现数据验证。

4 @log_results:日志输出

在运行复杂的数据分析时,跟踪每个函数的输出变得至关重要。@log_results装饰器可以帮助我们记录函数的结果,以便于调试和监控:
 def log_results(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        with open("results.log", "a") as log_file:
            log_file.write(f"{func.__name__} - Result: {result}\n")
        return result
    return wrapper
    
 @log_results
 def calculate_metrics(data):
    # Your metric calculation code here
将@log_results与日志库结合使用,以获得更高级的日志功能。

5 @suppress_errors:优雅的错误处理

数据科学项目经常会遇到意想不到的错误,可能会破坏整个计算流程。@suppress_errors装饰器可以优雅地处理异常并继续执行:
 def suppress_errors(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            print(f"Error in {func.__name__}: {e}")
            return None
    return wrapper
 
 @suppress_errors
 def preprocess_data(data):
    # Your data preprocessing code here
@suppress_errors可以避免隐藏严重错误,还可以进行错误的详细输出,便于调试。

6 @validate_output:确保质量结果

确保数据分析的质量至关重要。@validate_output装饰器可以帮助我们验证函数的输出,确保它在进一步处理之前符合特定的标准:

 def validate_output(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        if valid_output(result):
            return result
        else:
            raise ValueError("Invalid output. Please check your function logic.")
    return wrapper
 
 @validate_output
 def clean_data(data):
    # Your data cleaning code here
这样可以始终为验证函数输出定义明确的标准。

7 @retry:重试执行

@retry装饰器帮助我在遇到异常时重试函数执行,确保更大的弹性:

 import time
 
 def retry(max_attempts, delay):
    def decorator(func):
        def wrapper(*args, **kwargs):
            attempts = 0
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    print(f"Attempt {attempts + 1} failed. Retrying in {delay} seconds.")
                    attempts += 1
                    time.sleep(delay)
            raise Exception("Max retry attempts exceeded.")
        return wrapper
    return decorator
 
 @retry(max_attempts=3, delay=2)
 def fetch_data_from_api(api_url):
    # Your API data fetching code here
使用@retry时应避免过多的重试。

8 @visualize_results:漂亮的可视化

@visualize_results装饰器数据分析中自动生成漂亮的可视化结果:

 import matplotlib.pyplot as plt
 
 def visualize_results(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        plt.figure()
        # Your visualization code here
        plt.show()
        return result
    return wrapper
 
 @visualize_results
 def analyze_and_visualize(data):
    # Your combined analysis and visualization code here

9 @debug:调试变得更容易

调试复杂的代码可能非常耗时。@debug装饰器可以打印函数的输入参数和它们的值,以便于调试:
 def debug(func):
    def wrapper(*args, **kwargs):
        print(f"Debugging {func.__name__} start,- args: {args}, kwargs: {kwargs}")
        result = func(*args, **kwargs)
        print(f"Debugging {func.__name__} end,- return: {result}")
        return result
    return wrapper
    
 @debug
 def complex_data_processing(data, threshold=0.5):
    # Your complex data processing code here

10 @deprecated:处理废弃的函数

随着我们的项目更新迭代,一些函数可能会过时。@deprecated装饰器可以在一个函数不再被推荐时通知用户:
 import warnings
 
 def deprecated(func):
    def wrapper(*args, **kwargs):
        warnings.warn(f"{func.__name__} is deprecated and will be removed in future versions.", DeprecationWarning)
        return func(*args, **kwargs)
    return wrapper
 
 @deprecated
 def old_data_processing(data):
    # Your old data processing code here

作者:Gabe A, M.Sc

11 @log_decorator:日志装饰器

自动记录函数调用情况:

import logging

def log_decorator(func):
    def wrapper(*args, **kwargs):
        logging.info(f"调用函数 {func.__name__},参数: args={args}, kwargs={kwargs}")
        return func(*args, **kwargs)
    
    return wrapper

logging.basicConfig(level=logging.INFO)

@log_decorator
def process_data(data):
    # 假设这是一个处理数据的函数
    processed_data = data * 2
    return processed_data

# 使用装饰器记录process_data函数的调用情况
result = process_data(10)
print(result)

12 @permission_required:权限检查装饰器

用于验证用户是否有执行特定操作的权限:
def permission_required(permission_level):
    def decorator(func):
        def wrapper(user):
            if user.permission >= permission_level:
                return func(user)
            else:
                raise PermissionError(f"用户权限不足,需要{permission_level}级以上才能执行此操作")
        return wrapper
    return decorator

class User:
    def __init__(self, username, permission):
        self.username = username
        self.permission = permission

@permission_required(3)
def admin_only_operation(user):
    print(f"{user.username}正在执行管理员操作...")

# 测试权限装饰器
user1 = User("Alice", 2)
user2 = User("Bob", 3)

admin_only_operation(user1)  # 预期抛出PermissionError异常
admin_only_operation(user2)  # 应该正常执行

13 @lru_cache:缓存装饰器

用于提高函数计算结果的复用率:

from functools import lru_cache

@lru_cache(maxsize=100)
def expensive_computation(arg):
    # 假设这是个耗时较长的计算过程
    simulated_delay = arg * 0.1  # 模拟延迟时间
    time.sleep(simulated_delay)
    return arg * arg

# 使用缓存装饰器加速重复计算
print(expensive_computation(5))  # 第一次调用会计算并缓存结果
print(expensive_computation(5))  # 第二次调用直接返回缓存的结果

14 @transactional:事务处理装饰器

确保数据库操作的原子性:

class DatabaseWrapper:
    def __init__(self):
        self.transactions = []

    def begin_transaction(self):
        self.transactions.append(True)

    def commit(self):
        self.transactions.pop()

    def rollback(self):
        if self.transactions:
            self.transactions.pop()

def transactional(db):
    def decorator(func):
        def wrapper(*args, **kwargs):
            db.begin_transaction()
            try:
                result = func(db, *args, **kwargs)
                db.commit()
                return result
            except Exception as e:
                db.rollback()
                raise e from None
        return wrapper
    return decorator

@transactional
def update_user_profile(db, user_id, new_email):
    # 假设这里是更新用户资料的操作,涉及到数据库事务处理
    pass  # 实际操作代码省略

# 使用事务装饰器保证数据操作的安全性
db = DatabaseWrapper()
update_user_profile(db, 1, 'new@email.com')

15 @type_check_decorator:类型检查装饰器

确保函数传入参数类型正确:

def type_check_decorator(typespec):
    def decorator(func):
        def wrapper(*args, **kwargs):
            expected_args = typespec.get('args', [])
            expected_kwargs = typespec.get('kwargs', {})
            for i, (arg, expected_type) in enumerate(zip(args, expected_args)):
                if not isinstance(arg, expected_type):
                    raise TypeError(f"参数{i}的类型错误,期望{expected_type},但实际是{type(arg)}")
            for key, value in kwargs.items():
                if key in expected_kwargs and not isinstance(value, expected_kwargs[key]):
                    raise TypeError(f"关键字参数'{key}'的类型错误,期望{expected_kwargs[key]},但实际是{type(value)}")
            return func(*args, **kwargs)
        return wrapper
    return decorator

@type_check_decorator({'args': [int, str], 'kwargs': {'age': int}})
def greet_user(name, age=None):
    print(f"Hello, {name}! You're {age} years old.")

# 测试类型检查装饰器
greet_user("Alice", 25)  # 正确类型,应正常执行
greet_user("Alice", "twenty-five")  # 错误类型,应抛出TypeError异常

16 @ThrottleDecorator:性能优化装饰器

限制函数执行频率(节流):

import time
from threading import Lock

class ThrottleDecorator:
    def __init__(self, interval_seconds):
        self.interval = interval_seconds
        self.last_called = 0
        self.lock = Lock()

    def __call__(self, func):
        def wrapper(*args, **kwargs):
            with self.lock:
                now = time.time()
                if now - self.last_called > self.interval:
                    self.last_called = now
                    return func(*args, **kwargs)
                else:
                    print(f"函数{func.__name__}在{self.interval}秒内被调用过于频繁,本次调用被忽略.")
        return wrapper

@ThrottleDecorator(interval_seconds=3)
def send_request(url):
    # 假设这里是一个发送网络请求的函数
    print(f"向{url}发送请求...")

# 测试节流装饰器
for _ in range(10):
    send_request("http://example.com/api")

17 @profile_decorator:性能分析装饰器

使用cProfile模块分析函数运行性能:

import cProfile

def profile_decorator(func):
    def wrapper(*args, **kwargs):
        profile = cProfile.Profile()
        profile.enable()
        result = func(*args, **kwargs)
        profile.disable()
        profile.print_stats(sort='cumulative')
        return result
    return wrapper

@profile_decorator
def complex_algorithm(arr):
    # 假设这是一个复杂的算法实现
    pass  # 实际算法代码省略

data = [...]  # 输入数据
complex_algorithm(data)

18 @staticmethod和@classmethod:类静态方法和类方法装饰器

静态方法和类方法在面向对象编程中至关重要。它们允许定义在类本身而不是类的实例上调用的方法。

class MyClass:
    class_variable = "I am a class variable"
    
    @staticmethod
    def static_method():
        print("This is a static method")
        
    @classmethod
    def class_method(cls):
        print(f"This is a class method. Accessing class variable: {cls.class_variable}")

# Usage
MyClass.static_method()
MyClass.class_method()

19 @property:类属性装饰器

@property 装饰器允许定义像属性一样访问的方法,可以使代码更具可读性并保持封装性。

class Circle:
    def __init__(self, radius):
        self._radius = radius

    @property
    def radius(self):
        return self._radius
        
    @radius.setter
    def radius(self, value):
        if value < 0:
            raise ValueError("Radius cannot be negative")
        self._radius = value

# Usage
circle = Circle(5)
print(circle.radius)
circle.radius = 7

20 @singleton:单例模式装饰器

使用装饰器实现单例模式:

from functools import wraps

def singleton(orig_cls):
    orig_new = orig_cls.__new__
    instance = None
    
    @wraps(orig_cls.__new__)
    def __new__(cls, *args, **kwargs):
        nonlocal instance
        if instance is None:
            instance = orig_new(cls, *args, **kwargs)
        return instance
    
    orig_cls.__new__ = __new__
    return orig_cls

@singleton
class Logger:
    def log(msg):
        print(msg)

logger1 = Logger()
logger2 = Logger()
assert logger1 is logger2

在这个示例中,我们定义了一个名为singleton的装饰器函数,它接受一个类作为参数,并返回一个经过装饰后的类。通过在类定义前使用@singleton装饰器,我们可以将该类转换为单例模式。

装饰器的作用是改变装饰的类的实例化方式。每当请求创建一个新对象时,都会得到相同的对象。在上面的示例中,我们创建了两个Logger对象,并使用assert语句验证这两个变量实际上引用了同一个对象。

这种装饰器实现的单例模式基于Python创建类的方式。当创建一个新实例时,会调用类的特殊方法__new__来创建实例。装饰器通过重写类的__new__方法,使其每次调用时返回相同的实例,从而实现单例模式。


参考:

anzhihe 安志合个人博客,版权所有 丨 如未注明,均为原创 丨 转载请注明转自:https://chegva.com/5975.html | ☆★★每天进步一点点,加油!★★☆ | 

您可能还感兴趣的文章!

发表评论

电子邮件地址不会被公开。 必填项已用*标注