Python装饰器的应用场景代码总结

时间:2021-05-22

装饰器的应用场景

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

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

函数注册表

简单注册表

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

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

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.registerdef a(): return 3 @r2.registerdef b(): return 5 @r1.register@r2.register

执行时封装代码

类型检查

from functools import wrapsdef 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

用户验证

from functools import wrapsclass 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 Falsedef 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

输出格式化

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

异常捕获

import jsonfrom functools import wrapsclass 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_ouputdef error(): raise Error1("该条异常会被捕获并按JSON格式输出")

日志管理

import timeimport loggingfrom functools import wrapsdef 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)模块导入时立即执行

装饰类

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

import timefrom functools import wrapsdef 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 pymysqlclass 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.contextmanagerdef acceptable_error_codes(*codes): try: yield except ShellException as exc_instance: if exc_instance.code not in codes: raise pass

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。

声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。

相关文章