装饰器

无参装饰器

需求: 一个加法函数, 想增强它的功能,能够输出被调用过程以及调用的参数信息.

def add(x, y):

return x + y

或者增加信息输出功能:

def add(x, y):

print('call add, x + y') # 日志输出到控制台

return x + y

该函数已完成要求, 但有以下缺点:

打印语句的耦合太高;
加法函数属于业务功能, 而输出信息的功能,属于非业务功能, 不该放在业务函数加法中.

更改:

def add(x, y):

return x + y

def logger(fn):

print('begin')

x = fn(4, 5)

print('end')

return x

print(logger(add))

该函数做到了业务分离,但fn函数调用传参有问题.

进一步更改:

def add(x, y):

return x + y

def logger(fn, *args, **kwargs):

print('begin')

x = fn(*args, **kwargs)

print('end')

return x

print(logger(add, 5, y=60))

装饰器语法糖

示例:

def logger(fn):

def wrapper(*args, **kwargs):

print('begin')

x = fn(*args, **kwargs)

print('end')

return x

return wrapper

@logger # 等价于 add = logger(add)

def add(x, y):

return x + y

print(add(6, y=60))

@logger就是装饰器语法.

装饰器(无参):

是一个函数;

函数作为它的形参;

返回值也是一个函数;

使用@functionname方式,简化调用;

装饰器和高阶函数:

装饰器是高阶函数, 但装饰器是对传入函数的功能的装饰(即功能增强).

装饰器示例:

import datetime

import time

def logger(fn):

def wrap(*args, **kwargs):

print("args={}, kwargs={}".format(args,kwargs))

start = datetime.datetime.now()

ret = fn(*args, **kwargs)

duration = datetime.datetime.now() - start

print("function {} took {}s".format(fn.__name__, duration.total_seconds()))

return ret

return wrap

@logger # add = logger(add)

def add(x, y):

""" This a function of addition."""

print("===call add===========")

time.sleep(2)

return x + y

print(add(4, y=7))

"__name__" 属性用来查看函数名.

理解装饰器:

文档字符串

python的文档:

python是文档字符串Documentation Strings.

在函数语句块的第一行, 且习惯是多行的文本, 所以一般使用三引号.

惯例是首字母大写, 第一行写概述, 空一行, 第三行写详细描述.

可以使用特殊属性 __doc__ 访问这个文档.

def add(x,y):

"""This is a function of addition"""

a = x+y

return x + y

print("func_name = {} doc = {}".format(add.__name__, add.__doc__))

print(help(add))

副作用:

def func(fn):

"""This is a function of decorator"""

def wrapper(*args, **kwargs):

"""This is a function of wrapper"""

c = fn(*args, **kwargs)

return c

return wrapper

@func # add = func(add)

def add(x, y):

"""This is a function od addition."""

return x + y

print(add(4, 5))

print('name: {}, doc: {}'.format(add.__name__, add.__doc__))

# 运行结果:

name:wrapper

doc:This is a function of wrapper

原函数对象的属性都被替换了, 而使用装饰器, 需要查看被封装函数的属性.

提供一个函数, 被封装函数属性 – copy -> 包装函数属性.

def copy_properties(src,dst):

dst.__name__ = src.__name__

dst.__doc__ = src.__doc__

def logger(fn):

def wrapper(*args,**kwargs):

'''This is a wrapper'''

print('before')

ret = fn(*args,**kwargs) # add3(4,5,6,y=6,z=5)

print('after')

return ret

copy_properties(fn,wrapper)

return wrapper

@logger

def add(x,y):

'''

This is a function

return int

x int

y int

'''

ret = x + y

return ret

print(add(4, 5), add.__name__, add.__doc__, sep=' ')

通过copy_properties函数将被包装函数的属性覆盖掉包装函数;

凡是被装饰的函数都需要复制这些属性, 这个函数很通用;

可以将复制属性的函数构建成装饰器函数, 带参装饰器.

带参装饰器

def copy_properties(src):

def _copy(dst):

dst.__name__ = src.__name__

dst.__doc__ = src.__doc__

return dst

return _copy

def logger(fn):

@copy_properties(fn) # wrapper = copy_properties(fn)(wrapper)

def wrapper(*args,**kwargs):

'''This is a wrapper'''

print('before')

ret = fn(*args,**kwargs)

print('after')

return ret

return wrapper

@logger

def add(x,y):

'''

This is a function

return int

x int

y int

'''

ret = x + y

return ret

print(add(4, 5), add.__name__, add.__doc__, sep=' ')

带参装饰器:

获取函数的执行时长, 对时长超过阈值的函数记录一下.

import datetime, time

def copy(src):

def _copy(dst):

dst.__name__ = src.__name__

dst.__doc__ = src.__doc__

return dst

return _copy

def logger(duration):

def _logger(fn):

@copy(fn) # wrapper = copy(fn)(wrapper)

def wrapper(*args,**kwargs):

start = datetime.datetime.now()

ret = fn(*args,**kwargs)

delta = (datetime.datetime.now() - start).total_seconds()

print('so slow') if delta > duration else print('so fast')

return ret

return wrapper

return _logger

@logger(5) # add = logger(5)(add)

def add(x,y):

time.sleep(0.5)

return x + y

print(add(5, 6))

带参装饰器:

是一个函数;

函数作为它的形参;

返回值是一个不带参的装饰器函数;

使用@functionname(参数列表)方式调用;

可以看做在装饰器外层又加了一层函数.

将记录的功能提取出来,就可以通过外部提供的函数来灵活控制输出:

import datetime, time

def copy(src):

def _copy(dst):

dst.__name__ = src.__name__

dst.__doc__ = src.__doc__

return dst

return _copy

def logger(duration, func=lambda name, duration: print('{} took {}s'.format(name, duration))):

def _logger(fn):

@copy(fn) # wrapper = copy(fn)(wrapper)

def wrapper(*args,**kwargs):

start = datetime.datetime.now()

ret = fn(*args,**kwargs)

delta = (datetime.datetime.now() - start).total_seconds()

if delta < duration:

func(fn.__name__, duration)

return ret

return wrapper

return _logger

@logger(5)

def add(x,y):

time.sleep(1)

return x + y

print(add(5, 6))

functools模块 (1)

functools.update_wrapper(wrapper, wrapped, assigned=WRAPPER_ASSIGNMENTS, updated=WRAPPER_UPDATES)

类似copy_properties功能;

wrapper包装函数, wrapped被包装函数;

元组WRAPPER_ASSIGNMENTS中是需要被覆盖的属性;

'__module__', '__name__', '__qualname__', '__doc__', '__annotations__' (模块名, 名称, 限定名, 文档, 参数注解);

元组WRAPPER_UPDATES中是需要被更新的属性, __dict__属性字典;

增加一个__wrapped__属性, 保留着wrapped函数.

示例:

import datetime, time, functools

def logger(duration, func=lambda name, duration: print('{} took {}s'.format(name, duration))):

def _logger(fn):

def wrapper(*args,**kwargs):

start = datetime.datetime.now()

print(fn)

ret = fn(*args,**kwargs)

delta = (datetime.datetime.now() - start).total_seconds()

if delta > duration:

func(fn.__name__, duration)

return ret

return functools.update_wrapper(wrapper, fn) # 复制函数属性, 功能类似functools.wraps

return _logger

@logger(5) # add = logger(5)(add)

def add(x,y):

time.sleep(1)

return x + y

print(add(5, 6), add.__name__, add.__wrapped__, add.__dict__, sep=' ') # add.__wrapped与第7行有关.

使用functools.wraps复制属性:

import datetime, time, functools

def logger(duration, func=lambda name, duration: print('{} took {}s'.format(name, duration))):

def _logger(fn):

@functools.wraps(fn)

def wrapper(*args,**kwargs):

start = datetime.datetime.now()

print(fn)

ret = fn(*args,**kwargs)

delta = (datetime.datetime.now() - start).total_seconds()

if delta > duration:

func(fn.__name__, duration)

return ret

return wrapper

# return functools.update_wrapper(wrapper, fn) # 复制函数属性, 功能类似functools.wraps

return _logger

@logger(5) # add = logger(5)(add)

def add(x,y):

time.sleep(1)

return x + y

print(add(5, 6), add.__name__, add.__dict__, sep=' ')

参数注解 – Function Annotations

函数定义的弊端与解决办法

python是动态语言, 变量随时可以被赋值, 且能赋值为不同的类型;

python不是静态编译型语言, 变量类型是在运行器决定的;

动态语言很灵活, 但是也有弊端.

def add(x, y):

return x + y

print(add(5,4))

print(add('hello', 'world'))

print(add(4, 'hello')) # 报错, int不能和str直接相加.

# 难发现: 由于不做任何类型检查, 知道运行期问题才显现出来, 或者线上运行时才能暴露出问题;

# 难使用: 函数的使用者看到函数的时候, 并不知道你的函数设计, 并不知道应该传入什么类型的数据.

解决这种动态语言定义的弊端:

1.增加文档Documentation String. (__doc__)

惯例, 非强制标准.

函数定义更新与文档更新未必同步.

2.函数注解.

python3.5引入,

对函数的参数进行类型注解;

只对函数参数做一个辅助说明, 并不对函数参数进行类型检查.

提供第三方工具, 做代码分析, 发现隐藏的BUG.

函数注解信息,保存在 __annotations__属性中.

3.变量注解.

python3.6引入.

i:int = 3

函数注解:

In [1]: def add(x:int, y:int) -> int:

...: return x + y

...:

In [2]: print(add.__annotations__)

{'x': <class 'int'>, 'y': <class 'int'>, 'return': <class 'int'>}

In [3]:

业务应用

1.函数参数类型检查;

2.思路:

函数参数的检查,一定是在函数外.

函数应该作为参数,传入到检查函数中.

检查函数拿到函数传入的实际参数,与形参声明对比.

__annotations__属性是一个字典,其中包括返回值类型的声明.假设要做位置参数的判断,无法和字典中的声明对应,使用inspect模块.

3.inspect模块:

提供获取对象信息的函数,可以检查函数和类,类型检查.

inspect模块

signature(callable),获取签名

函数签名包含了一个函数的信息, 包括函数名,它的参数类型,类和名称空间及其他信息.

In [3]: import inspect

In [4]: def add(x:int, y:int, *args, **kwargs) -> int:

...: return x + y

...:

In [5]: sig = inspect.signature(add)

In [6]: sig

Out[6]: <Signature (x:int, y:int, *args, **kwargs) -> int>

In [7]: sig.parameters # OrderedDict

Out[7]:

mappingproxy({'args': <Parameter "*args">,

'kwargs': <Parameter "**kwargs">,

'x': <Parameter "x:int">,

'y': <Parameter "y:int">})

In [9]: sig.return_annotation

Out[9]: int

In [10]: sig.parameters['y']

Out[10]: <Parameter "y:int">

In [13]: sig.parameters['y'].annotation

Out[13]: int

In [15]: sig.parameters['args']

Out[15]: <Parameter "*args">

In [14]: sig.parameters['args'].annotation

Out[14]: inspect._empty

In [17]: sig.parameters['args'].annotation

Out[17]: inspect._empty

In [18]: sig.parameters['kwargs']

Out[18]: <Parameter "**kwargs">

In [19]: sig.parameters['kwargs'].annotation

Out[19]: inspect._empty

inspect.isfunction(add), 是否是函数;

isspect.ismethod(add), 是否是类的方法;

inspect.isgenerator(add), 是否是生成器对象;

inspect.isgeneratorfunction(add), 是否是生成器函数;

inspect.isclass(add), 是否是类;

inspect.ismodule(inspect), 是否是模块;

inspect.isbuiltin(print), 是否是内建函数.

...

还有很多is函数,需要的时候查阅inspect模块帮助.

parameter对象:

保存在元组中, 只读.

name为参数的名字.

annotation, 参数的注解, 可能没有定义.

default, 参数的缺省值, 可能没有定义.

empty, 特殊的类, 用来标记default属性或者注释annotation属性.

kind, 实参如何绑定到形参, 就是形参的类型.

POSITIONAL_ONLY, 值必须是位置参数提供;

POSITIONAL_OR_KEYWORD, 值可作为关键字或者位置参数提供.

VAR_POSITIONAL, 可变位置参数, 对应*args.

KEYWORD_ONLY, keyword-only参数, 对应*或者*args之后出现的非可变关键字参数.

VAR_KEYWORD, 可变关键字参数, 对应**kwargs.

举例:

import inspect

def add(x, y:int=7, *args, z, t=10,**kwargs) -> int:

return x + y

sig = inspect.signature(add)

print(sig)

print(sig.parameters)

for i, item in enumerate(sig.parameters.items()):

name, param = item

print('name: ',name, 'param: ',param)

print(i+1, name, param.annotation, param.kind, param.default)

print(param.default is param.empty, end=' ')

业务应用:

import inspect

def check(fn):

def wrapper(*args, **kwargs):

sig = inspect.signature(fn)

params = sig.parameters # OrderedDict

values = list(params.values())

for i,p in enumerate(args):

param = values[i]

if param.annotation is not param.empty and not isinstance(p, param.annotation):

print(p,'!==',values[i].annotation)

for k,v in kwargs.items():

if params[k].annotation is not inspect._empty and not isinstance(v, params[k].annotation):

print(k,v,'!===',params[k].annotation)

return fn(*args, **kwargs)

return wrapper

@check

def add(x, y:int=7) -> int:

return x + y

print(add(x=20, y=10))

functools模块(2)

partial方法

偏函数, 把函数部分的参数固定下来, 相当于为部分的参数添加了一个固定的默认值,形成一个新的函数并返回.

从partial生成得新函数,是对原函数的封装.

partial方法举例:

import functools

def add(x, y) -> int:

return x + y

newadd = functools.partial(add, y=5)

print(newadd(7))

print(newadd(7, y=6))

print(newadd(y=10, x=6))

import inspect

print(inspect.signature(newadd))

print('*' * 20)

import functools

def add(x, y, *args) -> int:

print(args)

return x + y

newadd = functools.partial(add, 1,3,6,5)

print(newadd(7))

print(newadd(7, 10))

# print(newadd(9, 10, y=20, x=26)) # 报错.

print(newadd())

import inspect

print(inspect.signature(newadd))

lru_cahce方法

@functools.lru_cache(maxsize=128, type=False)

least-recently-used装饰器. lru, 最近最少使用. cache缓存.

如果maxsize设置为None,则禁用LRU功能,并且缓存可无限制增长,当maxsize是二的幂时,LRU功能执行得最好.

如果typed设置为True,则不同类型的函数参数将单独缓存. 例:f(3)和f(3.0)将被视为具有不同结果的不同调用.

例:

import functools

import time

@functools.lru_cache()

def add(x, y, z=2):

time.sleep(z)

return(x+y)

print(add(4, 5))

print(add(4.0, 5))

print(add(4, 6))

print(add(4, 6, 3))

print(add(6, 4))

print(add(4, y=6))

print(add(x=4, y=6))

lru_cache装饰器:

通过一个字典缓存被装饰器函数的调用和返回值;

import functools

print(functools._make_key((4,6),{'z':3},False))

print(functools._make_key((4,6,3),{},False))

print(functools._make_key(tuple(),{'z':3,'x':4,'y':6},False))

print(functools._make_key(tuple(),{'z':3,'x':4,'y':6}, True))

斐波那契数列递归方法的改造:

import functools

@functools.lru_cache() # maxsize=None

def fib(n):

if n < 2:

return n

return fib(n-1) + fib(n-2)

print([fib(x) for x in range(35)])

lru_cache装饰器应用:

使用前提:

同样的函数参数一定得得到同样的结果;

函数执行时间很长,且要多次执行;

本质是函数调用的参数 -> 返回值.

缺点:

不支持缓存过期,key无法过期失效.

不支持清除操作.

不支持分布式,是一个单机的缓存.

适用场景, 单机上需要空间换时间的地方,可以用缓存来将计算变成快速的查询.