[toc]
python高阶函数和装饰器三
函数定义的弊端
- Python 是动态语言，变量随时可以被赋值，且能赋值为不同的类型;
- Python 不是表态编译型语言，变量类型是在运行时决定的;
- 动态语言很灵活，但是这种我也是弊端;
In [1]: def add(x,y): ...: return x + y In [2]: print(add(4,5)) 9 In [3]: print(add('hello','world')) helloworld In [4]: print(add(4,'world')) # 这时报错
- 难发现: 由于不做任何类型检查，直到运行期间问题才显现出来，或者线上运行时才能暴露出问题;
- 难使用: 函数的使用者看到函数的时候，并不知道你的函数设计，并不知道应该传入什么类型的数据;
- 函数注解(Function Annotations)
def add(x:int ,y:int) -> int: ''' :param x: :param y: :return: ''' return x + y print(add('ssjinyao','.com'))
- 函数注解
- Python3.5引入;
- 对函数的参数进行类型注解;
- 对函数的返回值进行类型注解;
- 只对函数参数做一个辅助的说，并不对函数参数进行类型检查;
- 提供给第三方工具，做代码分析，发现隐藏的bug;
- 函数注解的信息，保存在
__annotations__属性中
- 变量注解
Pyton3.6引入
i:int =3
业务应用
* 函数参数类型检查 * 思路
- 函数参数的检查，一定是在函数外;
- 函数应该作为参数，传入到检查函数中;
- 检查函数拿到函数传的实际参数，与形参声明对比;
__annotations__属jntg是一个字典，其中包括返回值类型的声明。假设要做位置参数的判断，无法和字典中的声明对应，使用inpsect 模块;inspect模块: 提供获取对象信息的函数，可以检查函数和类、类型检查;
inspect 模块
signature(callable)，获取签名(函数签名包含了一个函数的信息，包括函数名，它的参数类型、它所在的类和名称空间及其他信息)
import inspect def add(x:int,y:int,*args,**kwargs)->int: return x+y sig = inspect.signature(add) print(sig) print('params:',sig.parameters) # OrderedDict print('return:',sig.return_annotation) print(sig.parameters['y']) print(sig.parameters['x'].annotation) print(sig.parameters['args']) print(sig.parameters['args'].annotation) print(sig.parameters['kwargs']) print(sig.parameters['kwargs'].annotation) # 输出信息 (x: int, y: int, *args, **kwargs) -> int params: OrderedDict([('x', <Parameter "x: int">), ('y', <Parameter "y: int">), ('args', <Parameter "*args">), ('kwargs', <Parameter "**kwargs">)]) return: <class 'int'> y: int <class 'int'> *args <class 'inspect._empty'> **kwargs <class 'inspect._empty'>
inspect.isfunction(add) # 是否是函数 inspect.ismethod(add) # 是否是类的方法 inspect.isgenerator(add) # 是否是生成器对象 inspect.isgenratorfunction(add)# 是否是生成器函数 inspect.isclass(add) # 是否是类 inspect.ismodule(inspect) #是否是模块 inspect.isbuiltin(print) #是否是内建对象 # 还很多is函数，需要的时候查阅inpect模块帮助
- Parameter对象
- 保存在元组中，是只读的;
- name, 参数的名字;
- annotation，参数的注解，可能没有定义;
- default，参数的缺省值，可能没有定义;
- empty, 特殊的类，用来标记default属性或者注释annotation属性的空值;
kind，实参如何绑定到形参，就是形参的类型;
POSITIONAL_ONLY, 值必须是位置参数提供;
VAR_POSITIONAL, 可变位置参数,对应*args;
KEKWORD_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('params: ', sig.parameters) print('return: ', sig.return_annotation) for i,(name,param) in enumerate(sig.parameters.items()): print(i+1, name, param.annotation, param.kind, param.default) print(param.default is param.empty, end = '\n') # 输出 1 x <class 'inspect._empty'> POSITIONAL_OR_KEYWORD <class 'inspect._empty'> True 2 y <class 'int'> POSITIONAL_OR_KEYWORD 7 False 3 args <class 'inspect._empty'> VAR_POSITIONAL <class 'inspect._empty'> True 4 z <class 'inspect._empty'> KEYWORD_ONLY <class 'inspect._empty'> True 5 t <class 'inspect._empty'> KEYWORD_ONLY 10 False 6 kwargs <class 'inspect._empty'> VAR_KEYWORD <class 'inspect._empty'> True
import inspect def check(fn): def wrapper(*args,**kwargs): # 实参检查 print(args,kwargs) sig = inspect.signature(fn) print(sig) print('params: ', sig.parameters) print('return: ', sig.return_annotation) print('-------------------------') for param in sig.parameters.values(): print(param.name,param) print(param.name, param.annotation, param.kind, param.default) print(param.default is param.empty, end='\n') ret = fn(*args, **kwargs) return ret return wrapper @check def add(x:int , y:int=7) -> int: print(x + y) add(4,y=8) # 输出 (4,) {'y': 8} (x: int, y: int = 7) -> int params: OrderedDict([('x', <Parameter "x: int">), ('y', <Parameter "y: int = 7">)]) return: <class 'int'> ------------------------- x x: int x <class 'int'> POSITIONAL_OR_KEYWORD <class 'inspect._empty'> True y y: int = 7 y <class 'int'> POSITIONAL_OR_KEYWORD 7 False 12
import inspect from functools import wraps def check(fn): @wraps(fn) def wrapper(*args, **kwargs): # 实参检查 print(args,kwargs) sig = inspect.signature(fn) params = sig.parameters # 有序字典 param_list = tuple(params.keys()) for i, v in enumerate(args): k = param_list[i] if isinstance(v,params[k].annotation): print(v,'is',params[k].annotation) else: errstr = "{} {} {}".format(v,'is not',params[k].annotation) print(errstr) raise TypeError(errstr) # 关键字传参处理 for k, v in kwargs.items(): if isinstance(v,params[k].annotation): print(v,'is',params[k].annotation) else: errstr = "{} {} {}".format(v, "is not", params[k].annotation) print(errstr) raise TypeError(errstr) ret = fn(*args,**kwargs) return ret return wrapper @check def add(x: int, y:int = 7) -> int: print(x + y) return x + y add(4,y=10) # 输出 (4,) {'y': 10} 4 is <class 'int'> 10 is <class 'int'> 14
functools 模块
- partial方法
比较有用函数，尤其是在做函数封装的时候
- 偏函数，把函数的部分的参数固定下来，相当于部分的参数添加了一个固定的默认值，形成一个新的函数返回;
- 从partial 生成的新函数，是对函数的封装;
import functools def add(x, y:int)-> int: ret = x + y print(ret) return ret import inspect print(inspect.signature(add)) newadd = functools.partial(add,y=6) newadd(4) # 输出 (x, y: int) -> int 10
@functools.lru_cache(maxsize=128, typed=False)
- Least-recently-used 装饰器。 lru，最近最少使用。cache缓存;
- 如果maxsize设置为None，则禁用LRU功能，并且缓存可以无限制增长。当maxsize 是二的幂时，LRU功能执行得最好;
- 如果typed 设置为True，则不同类型的函数参数将单独缓存。例如,
f(3)和f(3.0)将被视为具有不同结果的不同调用。
测试一下不同函数参数的不同第二次调用速度
In [8]: import functools In [9]: import time In [13]: @functools.lru_cache() ...: def add(x,y=5): ...: time.sleep(3) ...: ret = x + y ...: print(ret) ...: return ret In [14]: add(4) 9 Out[14]: 9 In [15]: add(4) Out[15]: 9 In [16]: add(4,5) 9 Out[16]: 9 In [17]: add(4,y=5) 9 Out[17]: 9 In [18]: add(x=4,y=5) 9 Out[18]: 9
- lru_cache 装饰器
- 通过一个字典缓存被装饰函数的调用和返回值;
tuple 转 list 补充方法
In [19]: lst = [] In [20]: lst[:] = (1,2,3,4,5) In [21]: lst Out[21]: [1, 2, 3, 4, 5]
- 斐波那契数列递归方法的改造
import functools @functools.lru_cache() 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 无法过期、失效;
不支持清除操作;
不支持分布式，是一个单机的缓存; - 适用场景，单机上需要空间换时间的地方，可以用缓存来将计算变成快速的查询;
装饰器练习
- 实现一个cache装饰器，实现可过期，可删除的功能。可以不换出;
简化设计，函数的形参定义不包含可变位置、可变关键词参数和keyword-only参数可以不考虑缓存满了之后的换出问题;
数据类型的选择:
缓存的应用场景，是有数据需要频繁查询，且每次查询都需要大量计算或者等待时间之后才能返回的结果的情况，使用缓存来提高查询速度，用空间换取时间;
cache 应用选用什么数据结构？
便于查询的，且能快速找到数据的数据结构;
每次查询的时候，只要输入致，就应该得到同样的结果(顺序一致，例如减法函数，参数顺序不一致，结果不一样);
基于上面的分析，些数据结构应该是字典;
通过一个key, 对应一个value;
key是参数列表组成的结构，value 是函数返回值。难点在于key如何处理。
key的存储;
key 必须是hash类;
key能接受到位置参数和关键字参数传参;
位置参数是被收集在一个tuple中的，本身就有顺序;
key 的要求;
key必须是hashable;
由于key是所有实参组合而成的, 而且最好要作为key的, key一定要可以hash, 但是如果key有不可hash类型数据，就无法完成;
缓存必须使用key，但是key必须可hash,所以只能适用实参是不可变类型的函数调用;
key 算法设计
inspect模块获取函数签名后，取parameters，这是一个有序字典，会保存所有参数的信息;
构建一个字典params_dict，按照位置顺序从args中依次对应参数名和传入的实参，组成kv对，存入params_dict中;
kwargs所有值update到parmas_dict中;
如果使用了缺省的参数，不会出现实参params_dict中，会出现在签名的取parameters中,缺省值也在定义中;
from functools import wraps import time import inspect import datetime def m_cache(fn): local_cache = {} @wraps(fn) def wrapper(*args, **kwargs): print(args, kwargs) key_dict = {} #sorted sig = inspect.signature(fn) params = sig.parameters #有序字典 param_list = list(params.keys()) # 位置参数 for i,v in enumerate(args): # print(i,v) k = param_list[i] key_dict[k] =v # 关键字参数 # for k,v in kwargs.items(): # key_dict[k] = v key_dict.update(kwargs) # 缺省值的处理 for k in params.keys(): if k not in key_dict.keys(): key_dict[k] = params[k].default key = tuple(sorted(key_dict.items())) if key not in local_cache.keys(): ret = fn(*args, **kwargs) local_cache[key] = ret return local_cache[key] return wrapper def logger(fn): @wraps(fn) def wrapper(*args,**kwargs): start = datetime.datetime.now() ret = fn(*args,**kwargs) dalta = (datetime.datetime.now() - start).total_seconds() print(dalta) return ret return wrapper @logger @m_cache def add(x,y=5): time.sleep(3) ret = x + y return ret add(4) add(4,5) add(4,y=5) add(x=4,y=5) add(y=5,x=4) ## 对比输出结果，缓存输执行的效率 (4,) {} 3.001863 (4, 5) {} 0.000151 (4,) {'y': 5} 8.1e-05 () {'x': 4, 'y': 5} 7.6e-05 () {'y': 5, 'x': 4} 7.3e-05
过期功能;
一般缓存系统都有过期功能;
带参装饰器及格离化、函数封装
from functools import wraps import time import inspect import datetime def m_cache(duration): def _m_cache(fn): local_cache = {} @wraps(fn) def wrapper(*args, **kwargs): # local_cache 有没有过期的key def clear_expire(cache): expire_keys = [] for k, (_, ts) in cache.items(): if datetime.datetime.now().timestamp() - ts > duration: expire_keys.append(k) for k in expire_keys: cache.pop(k) print(args, kwargs) clear_expire(local_cache) def make_key(): key_dict = {} #sorted sig = inspect.signature(fn) params = sig.parameters #有序字典 param_list = list(params.keys()) # 位置参数 for i,v in enumerate(args): # print(i,v) k = param_list[i] key_dict[k] =v # 关键字参数 # for k,v in kwargs.items(): # key_dict[k] = v key_dict.update(kwargs) # 缺省值的处理 for k in params.keys(): if k not in key_dict.keys(): key_dict[k] = params[k].default return tuple(sorted(key_dict.items())) key = make_key() if key not in local_cache.keys(): ret = fn(*args, **kwargs) local_cache[key] = (ret, datetime.datetime.now().timestamp()) # (,,) = key, tuple => value # local_cache[key] = ret return local_cache[key] return wrapper return _m_cache def logger(fn): @wraps(fn) def wrapper(*args,**kwargs): start = datetime.datetime.now() ret = fn(*args,**kwargs) dalta = (datetime.datetime.now() - start).total_seconds() print(dalta) return ret return wrapper @logger @m_cache(6) def add(x,y=5): time.sleep(3) ret = x + y return ret add(4) add(4,5) add(4,y=5) add(x=4,y=5) add(y=5,x=4) time.sleep(6) add(4) add(4,5) add(4,y=5) add(x=4,y=5) add(y=5,x=4) ## 调试清除缓存功能 (4,) {} 3.003023 (4, 5) {} 0.000211 (4,) {'y': 5} 0.000121 () {'x': 4, 'y': 5} 0.000418 () {'y': 5, 'x': 4} 0.000123 (4,) {} 3.003332 (4, 5) {} 0.000277 (4,) {'y': 5} 0.000198 () {'x': 4, 'y': 5} 0.000182 () {'y': 5, 'x': 4} 0.000179
补充: 过期是什么 ？
它是某一个key过期，可以对每个key单独设置过期时间，也可以对这些key统一设定过期时间;
本次的实现就简单点，统一设定key的过期时间，当key生存超过了这个时间，就自动被清除;
- 写一个命令分发器
程序员可以方便的注册函数到某一个命令，用户输入命令时，路由到注册的函数;
如果此命令没有对应的注册函数，执行默认函数;
用户输入用input(“>>”)
代码实现
def cmds_dispatcher(): #命令和函数存储的地方 commands = {} # 注册 def reg(name): def _reg(fn): commands[name] =fn return _reg def defaultfunc(): print("Unknown command") def dispatcher(): while True: cmd = input(">> ") if cmd.strip() == 'quit': return commands.get(cmd,defaultfunc)() return reg,dispatcher cr,cp = cmds_dispatcher() # 自定久函数，注册 @cr('mag') def foo1(): print("welcome ssjinyao") @cr('py') def foo2(): print("welcome python") cp()
装饰器的用途
装饰器是AOP面向切面编程 Aspect Oriented Programming的思想的体现;
面向对象往往需要通过继承或者组合依赖等方式调用一些功能，这些功能的代码往往可能在多个类中出现，例如logger。这样造成代码的重复，增加了耦合。logger的改变影响所有使用它的类或方法。
而AOP在需要的类或方法上切下，前后的切入点可以加入增强的功能。让调用者和被调用者解耦。
这是一种不修改原来的业务代码，给程序动态加功能的技术。例如logger函数功能就是对业务函数增加日志的，而业务函数中应该把业务无关的日志功能剥离干净。
装饰器应用场景
日志、监控、权限、设计、参数检查、路由等处理；
这些功能与业务功能无关，很多业务都需要的公共功能，所以适合独立出来，需要的时候，对目标对象增强。
格言
不应以圣贤自命、以小人目人、面色如铁、话语如刀
