Python装饰器的应用场景

╰+攻爆jí腚メ 2023-10-09 05:54 166阅读 0赞

### 装饰器的应用场景 ###

*  附加功能
 *  数据的清理或添加:
    
     *  函数参数类型验证 @require\_ints 类似请求前拦截
     *  数据格式转换 将函数返回字典改为 JSON/YAML 类似响应后篡改
     *  为函数提供额外的数据 mock.patch
 *  函数注册
    
     *  在任务中心注册一个任务
     *  注册一个带信号处理器的函数

### 不同应用场景下装饰器实现 ###

#### 函数注册表 ####

1.  简单注册表

funcs = []
    def register(func):
        funcs.append(func)
        return func
        
        
    @register
    def a():
        return 3
        
    @register
    def b():
        return 5
        
    
    # 访问结果
    result = [func() for func in funcs]

1.  注册表隔离(使用类的不同实例)

class Registry(object):
        def __init__(self):
            self._funcs = []
        
        def register(self, func):
            self._funcs.append(func)
            
        def run_all(self):
            return [func() for func in self._funcs]
            
            
    r1 = Registry()
    r2 = Registry()
    
    @r1.register
    def a():
        return 3
        
    @r2.register
    def b():
        return 5
        
    @r1.register
    @r2.register

#### 执行时封装代码 ####

1.  类型检查

from functools import wraps
    
    def require_ints(func):
        @wraps(func)  # 将func的信息复制给inner
        def inner(*args, **kwargs):
            for arg list(args) + list(kwargs.values()):
                if not isinstance(arg, int:
                    raise TypeError("{} 只接受int类型参数".format(func.__name__)
            return func(*args, **kwargs)
        return inner

1.  用户验证

from functools import wraps
    
    class User(object):
        def __init__(self, username, email):
            self.username = username
            self.email = email
            
    class AnonymousUser(object):
        def __init__(self):
            self.username = self.email = None
            
        def __nonzero__(self):  # 将对象转换为bool类型时调用
            return False
            
    def requires_user(func):
        @wraps(func)
        def inner(user, *args, **kwargs):  # 由于第一个参数无法支持self, 该装饰器不支持装饰类
            if user and isinstance(user, User):
                return func(use, *args, **kwargs)
            else:
                raise ValueError("非合法用户")
        return inner

1.  输出格式化

import json
    from functools import wraps
    
    def json_output(func):  # 将原本func返回的字典格式转为返回json字符串格式
        @wrap(func)
        def inner(*args, **kwargs):
            return json.dumps(func(*args, **kwargs))
        return inner

1.  异常捕获

import json
    from functools import wraps
    
    class Error1(Exception):
        def __init__(self, msg):
            self.msg = msg
        def __str__(self):
            return self.msg
            
    
    def json_output(func):
        @wrap(func)
        def inner(*args, **kwargs):
            try:
                result = func(*args, **kwargs)
            except Error1 as ex:
                result = {"status": "error", "msg": str(ex)}
            return json.dumps(result)
        return inner
    
    # 使用方法
    @json_ouput
    def error():
        raise Error1("该条异常会被捕获并按JSON格式输出")

1.  日志管理

import time
    import logging
    from functools import wraps
    
    def logged(func):
        @wraps(func)
        def inner(*args, **kwargs):  # *args可以装饰函数也可以装饰类
            start = time.time()
            result = func(*args, **kwargs)
            exec_time = time.time() - start
            logger = logging.getLoger("func.logged")
            logger.warning("{} 调用时间:{:.2} 执行时间:{:.2}s 结果:{}".format(func.__name__, start, exec_time, result)

#### 带参数的装饰器 ####

带参数的装饰器相当于一个返回装饰器的函数,`@deco(a=1)`在调用`@`之前会首先执行`deco(a=1)`得到一个实际的装饰器, 带参数的装饰器`deco(a=1)`模块导入时立即执行

### 装饰类 ###

1.  为类增加可排序功能(而不通过继承子类扩充父类方法,比如多个类需要增加此功能时)

import time
    from functools import wraps
    
    def sortable_by_created(cls):
        original_init = cls.__init__
        
        @wrap(original_init)
        def new_init(self, *args, **kwargs):
            original_init(*args, **kwargs)
            self._created = time.time()
        
        cls.__init__ = new_init
        
        cls.__lt__ = lambda self, other: self._created < other._created
        cls.__gt__ = lambda self, other: self._created > other._created
        return cls

> 也可定义一个SortableByCreated()类, 子类使用多重继承其父类和SortableByCreated

### 类型转换 ###

函数被装饰后有可能变为一个类的实例,此时为了兼容函数调用,应为所返回的类提供`__call__`方法

class Task(object):
        def __call__(self, *args, **kwargs):
            return self.run(*args, **kwargs)
        def run(self, *args, **kwargs):
            raise NotImplementedError("子类未实现该接口")
            
    def task(func):
        class SubTask(Task):
            def run(self, *args, **kwargs):
                func(*args, **kwargs)
        return SubTask()

## 第二章 上下文管理器 ##

### 定义 ###

*  包装任意代码
 *  确保执行的一致性

### 语法 ###

*  with语句
 *  \_\_enter\_\_和\_\_exit\_\_方法

class ContextManager(object):
        def __init__(self):
            self.entered = False
            
        def __enter__(self):
            self.entered = True
            return self
            
        def __exit__(self, exc_type, exc_instance, traceback):
            self.entered = False

### 应用场景 ###

#### 资源清理 ####

import pymysql
    
    class DBConnection(object):
        def __init__(self, *args, **kwargs):
            self.args,self.kwargs = args, kwargs
            
        def __enter__(self):
            self.conn = pymysql.connect(*args, **kwargs)
            return self.conn.cursor()
            
        def __exit__(self, exc_type, exc_instance, trackback):
            self.conn.close()

#### 异常处理(避免重复) ####

*  传播异常(\_\_exit\_\_中return False)
 *  终止异常(\_\_exit\_\_中return True)

class BubleExceptions(object):
        def __enter__(self):
            return self
        def __exit__(self, exc_type, exc_instance, trackback):
            if exc_instance:
                print("出现异常: {}".format(exc_instance)
            return False   # return True终止异常

*  处理特定的异常

class HandleValueError(object):
        def __enter__(self):
            return self
        def __exit__(self, exc_type, exc_instance, trackback):
            if not exc_type: return True
            if issubclass(exc_type, ValueError): 
                print("处理ValueError: {}".format(exc_instance)
            return False

> `if issubclass...`语句改为`if exec_type == ValueError`则不处理ValueType的子类异常

> 也可以根据异常的属性来判断是否传播或终止

### 更简单的语法 ###

import contextlib
    
    @contextlib.contextmanager
    def acceptable_error_codes(*codes):
        try:
            yield
        except ShellException as exc_instance:
            if exc_instance.code not in codes:
                raise
            pass

转载于:https://www.cnblogs.com/superhin/p/11454823.html