Python装饰器的应用场景代码总结
装饰器的应用场景
- 附加功能
- 数据的清理或添加:
- 函数参数类型验证@require_ints类似请求前拦截
- 数据格式转换将函数返回字典改为JSON/YAML类似响应后篡改
- 为函数提供额外的数据mock.patch
- 函数注册
- 在任务中心注册一个任务
- 注册一个带信号处理器的函数
不同应用场景下装饰器实现
函数注册表
简单注册表
funcs=[] defregister(func): funcs.append(func) returnfunc @register defa(): return3 @register defb(): return5 #访问结果 result=[func()forfuncinfuncs]
注册表隔离(使用类的不同实例)
classRegistry(object): def__init__(self): self._funcs=[] defregister(self,func): self._funcs.append(func) defrun_all(self): return[func()forfuncinself._funcs] r1=Registry() r2=Registry() @r1.register defa(): return3 @r2.register defb(): return5 @r1.register @r2.register
执行时封装代码
类型检查
fromfunctoolsimportwraps
defrequire_ints(func):
@wraps(func)#将func的信息复制给inner
definner(*args,**kwargs):
forarglist(args)+list(kwargs.values()):
ifnotisinstance(arg,int:
raiseTypeError("{}只接受int类型参数".format(func.__name__)
returnfunc(*args,**kwargs)
returninner
用户验证
fromfunctoolsimportwraps
classUser(object):
def__init__(self,username,email):
self.username=username
self.email=email
classAnonymousUser(object):
def__init__(self):
self.username=self.email=None
def__nonzero__(self):#将对象转换为bool类型时调用
returnFalse
defrequires_user(func):
@wraps(func)
definner(user,*args,**kwargs):#由于第一个参数无法支持self,该装饰器不支持装饰类
ifuserandisinstance(user,User):
returnfunc(use,*args,**kwargs)
else:
raiseValueError("非合法用户")
returninner
输出格式化
importjson fromfunctoolsimportwraps defjson_output(func):#将原本func返回的字典格式转为返回json字符串格式 @wrap(func) definner(*args,**kwargs): returnjson.dumps(func(*args,**kwargs)) returninner
异常捕获
importjson
fromfunctoolsimportwraps
classError1(Exception):
def__init__(self,msg):
self.msg=msg
def__str__(self):
returnself.msg
defjson_output(func):
@wrap(func)
definner(*args,**kwargs):
try:
result=func(*args,**kwargs)
exceptError1asex:
result={"status":"error","msg":str(ex)}
returnjson.dumps(result)
returninner
#使用方法
@json_ouput
deferror():
raiseError1("该条异常会被捕获并按JSON格式输出")
日志管理
importtime
importlogging
fromfunctoolsimportwraps
deflogged(func):
@wraps(func)
definner(*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)模块导入时立即执行
装饰类
为类增加可排序功能(而不通过继承子类扩充父类方法,比如多个类需要增加此功能时)
importtime fromfunctoolsimportwraps defsortable_by_created(cls): original_init=cls.__init__ @wrap(original_init) defnew_init(self,*args,**kwargs): original_init(*args,**kwargs) self._created=time.time() cls.__init__=new_init cls.__lt__=lambdaself,other:self._createdother._created returncls
也可定义一个SortableByCreated()类,子类使用多重继承其父类和SortableByCreated
类型转换
函数被装饰后有可能变为一个类的实例,此时为了兼容函数调用,应为所返回的类提供__call__方法
classTask(object):
def__call__(self,*args,**kwargs):
returnself.run(*args,**kwargs)
defrun(self,*args,**kwargs):
raiseNotImplementedError("子类未实现该接口")
deftask(func):
classSubTask(Task):
defrun(self,*args,**kwargs):
func(*args,**kwargs)
returnSubTask()
第二章上下文管理器
定义
包装任意代码
确保执行的一致性
语法
with语句
__enter__和__exit__方法
classContextManager(object): def__init__(self): self.entered=False def__enter__(self): self.entered=True returnself def__exit__(self,exc_type,exc_instance,traceback): self.entered=False
应用场景
资源清理
importpymysql classDBConnection(object): def__init__(self,*args,**kwargs): self.args,self.kwargs=args,kwargs def__enter__(self): self.conn=pymysql.connect(*args,**kwargs) returnself.conn.cursor() def__exit__(self,exc_type,exc_instance,trackback): self.conn.close()
异常处理(避免重复)
传播异常(__exit__中returnFalse)
终止异常(__exit__中returnTrue)
classBubleExceptions(object):
def__enter__(self):
returnself
def__exit__(self,exc_type,exc_instance,trackback):
ifexc_instance:
print("出现异常:{}".format(exc_instance)
returnFalse#returnTrue终止异常
处理特定的异常
classHandleValueError(object):
def__enter__(self):
returnself
def__exit__(self,exc_type,exc_instance,trackback):
ifnotexc_type:returnTrue
ifissubclass(exc_type,ValueError):
print("处理ValueError:{}".format(exc_instance)
returnFalse
ifissubclass...语句改为ifexec_type==ValueError则不处理ValueType的子类异常
也可以根据异常的属性来判断是否传播或终止
更简单的语法
importcontextlib @contextlib.contextmanager defacceptable_error_codes(*codes): try: yield exceptShellExceptionasexc_instance: ifexc_instance.codenotincodes: raise pass
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。