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

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

简单注册表 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]

注册表隔离(使用类的不同实例) 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

执行时封装代码

类型检查 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

用户验证 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

输出格式化 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

异常捕获 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格式输出")

日志管理 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)模块导入时立即执行

装饰类

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

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