Python进阶3---python类型注解、functools

函数定义的弊端

函数注解Function Annotations

业务应用

inspect模块

#示例
import inspect
def add(x,y:int,*args,**kwargs) -> int:
    return x+y

sig = inspect.signature(add)
print('sig:',sig)
print('params :',sig.parameters)#Ordereddict
print('return :',sig.return_annotation)
print(sig.parameters['x'])
print(sig.parameters['y'])
print(sig.parameters['y'].annotation)
print(sig.parameters['args'])
print(sig.parameters['args'].annotation)
print(sig.parameters['kwargs'])
print(sig.parameters['kwargs'].annotation)
print(sig.parameters['kwargs'].empty)
print(sig.parameters['kwargs'].kind)
'''
sig: (x, y:int, *args, **kwargs) -> int
params : OrderedDict([('x', <Parameter "x">), ('y', <Parameter "y:int">), ('args', <Parameter "*args">), ('kwargs', <Parameter "**kwargs">)])
return : <class 'int'>
x
y:int
<class 'int'>
*args
<class 'inspect._empty'>
**kwargs
<class 'inspect._empty'>
<class 'inspect._empty'>
VAR_KEYWORD
'''

业务应用

 

functools模块进阶

 

 

 

#源码 _make_key _HashSeq
def _make_key(args, kwds, typed,
             kwd_mark = (object(),),
             fasttypes = {int, str, frozenset, type(None)},
             tuple=tuple, type=type, len=len):

    key = args
    if kwds:
        key += kwd_mark
        for item in kwds.items():
            key += item
    if typed:
        key += tuple(type(v) for v in args)
        if kwds:
            key += tuple(type(v) for v in kwds.values())
    elif len(key) == 1 and type(key[0]) in fasttypes:
        return key[0]
    return _HashedSeq(key)

class _HashedSeq(list):
    __slots__ = 'hashvalue'

    def __init__(self, tup, hash=hash):
        self[:] = tup
        self.hashvalue = hash(tup)

    def __hash__(self):
        return self.hashvalue

#利用缓存可以大大提高效率,类似于用空间换取时间!注意缓存与缓冲的区别!
import functools
@functools.lru_cache()
def fib(n):
    if n<2:
        return n
    else:return fib(n-1)+fib(n-2)

print(fib(100))

 装饰器练习*****

#第一题
#装饰器的应用
#实现一个cache装饰器,实现可过期的被清除的功能
'''
简化设计,函数的形参定义不包括可变位置参数、可变关键词参数和keyword-only参数
可以不考虑缓存满了之后的换出问题。

数据类型的选择:
缓存的应用场景,是有数据需要频繁查询,且每次查询都需要大量计算或等待时间之后才能返回结果的情况,
使用缓存来提高查询速度,用空间换取时间。

cache应该选用什么数据结构?
便于查询的,且能快速找到的数据结构。每次查询的时候,只要输入一致,就应该得到同样的结果(顺序也一致,例如减法函数,参数顺序不一致,结果不一样)
基于上面的分析,此数据结构应该是字典。通过一个key,对应一个value。
key是参数列表组成的结构,value是函数返回值。难点在于key如何处理!

key的存储
key必须是hash类,key能接受位置参数和关键字参数传参,位置参数是被收集在一个tuple中的,本身就有序
关键字参数被收集在一个字典中,本身无序,这会带来一个问题,传参的顺序未必是字典中保存的顺序。如何解决?
OrderedDict行吗?可以,它可以记录顺序。
不用OrderedDict行吗?可以,用一个tuple保存排过序的字典的item的kv对。

key的异同
什么才算是相同的key呢?定义一个加法函数,那么传参方式就应该有以下4种:
1.add(4,5)
2.add(4,y=5)
3.add(x=4,y=5)
4.add(y=5,x=4)
上面4种,可以有下面两种理解:
第一种: 3和4相同,1,2和3都不同。
第二种: 1、2、3、4全部相同。
lru_cache实现了第一种,可以看出单独的处理了位置参数和关键字参数。
但是函数定义为def add(4,y=5),使用了默认值,如何理解add(4,5)和add(4)是否一样呢?
如果认为一样,那么lru_cache无能为力,就需要使用inspect来自己实现算法。

key的要求
key必须是hashable,由于key是所有实参组合而成,而且最好要作为key的,key一定要可以hash,但是如果key有不可hash类型数据,就无法完成。
lru_cache就不可以,缓存必须使用key,但是key必须可hash,所以只能适用实参是不可变类型的函数调用。

key算法设计
inspect模块获取函数签名后,取parameters,这是一个有序字典,会保存所有参数的信息。
构建一个字典params_dict,按照位置顺序从args中依次对应参数名和传入的实参,组成kv对,存入params_dict中。
kwargs所有值update到params_dict中。
如果使用了缺省值的参数,不会出现在实参params_dict中,会出现在签名parameters中,缺省值也定义在其中。

调用的方式
普通的函数调用可以,但是过于明显,最好类似lru_cache的方式,让调查者无察觉的使用缓存。
构建装饰器函数

过期功能
一般缓存系统都有过期功能。
过期什么呢?
它是某一个key过期。可以对每一个key单独设置过期时间,也可以对这些key统一设定过期时间。
本次的实现就简单点,统一设置key的过期时间,当key生存超过了这个时间,就自动被清除。
注意:这里并没有考虑多线程等问题。而且这种过期机制,每一次都有遍历所有数据,大量数据的时候,
遍历可能有效率问题。
在下面的装饰器中增加一个参数,需要用到了带参装饰器了。
@mag_cache(5)代表key生存5秒钟后过期。
带参装饰器相当于在原来的装饰器外面再嵌套一层。

清除的时机,何时清除过期key?
1. 用到某个key之前,先判断是否过期,如果过期重新调用函数生成新的key对应value值。
2.一个线程负责清除过期的key,这个后面实现。本次在创建key之前,清除所有过期的key。

value的设计
1.key =>(v,createtimestamp)
适合key过期的时间都是统一的设定
2.key =>(v,createtimestamp,duration)
duration是过期时间,这样每一个key就可以单独控制过期时间。在这种设计中,-1可以表示永不过期,
0可以表示立即过期,正整数表示持续一段时间过期。
这次采用第一种实现!
'''
 1 #题目:实现一个cache装饰器,实现可过期、可自动清除的功能。
 2 
 3 from functools import wraps
 4 import time
 5 import inspect
 6 import datetime
 7 
 8 def cache(duration):
 9     def _cache(fn,):
10         local_cache = {}
11         @wraps(fn)
12         def wrapper(*args,**kwargs):
13             #local_cache中有没有过期的key
14             # for k,(_,ts) in local_cache:
15             #     if datetime.datetime.now().timestamp()- ts>5:
16             #         local_cache.pop(k)  #不能一边迭代一边删除!
17             def clear_expire(cache):
18                 expire_key = []
19                 for k, (_, ts) in cache.items():
20                     if datetime.datetime.now().timestamp() - ts > duration:
21                         expire_key.append(k)
22                 for k in expire_key:
23                     cache.pop(k)
24             clear_expire(local_cache)
25 
26             def make_key():
27                 key_dict = {}
28                 sig = inspect.signature(fn)
29                 params = sig.parameters  # 返回一个有序字典
30                 params_list = list(params.keys())  # 返回参数列表
31                 # 位置参数add(5,6,y=8)
32                 for i, v in enumerate(args):
33                     k = params_list[i]
34                     key_dict[k] = v
35                 # 关键字参数
36                 # print('*kwargs', *kwargs)
37                 # for k,v in kwargs.items():
38                 #     key_dict[k] = v
39                 key_dict.update(kwargs)
40                 # 缺省值参数
41                 for k in params.keys():
42                     if k not in key_dict.keys():
43                         key_dict[k] = params[k].default
44                 key = tuple(sorted(key_dict.items()))
45                 # print('处理后得到的key:',key)
46                 # print('处理后得到的local_cache:',local_cache)
47                 return key
48 
49             key = make_key()
50 
51             if key not in local_cache.keys():
52                 ret = fn(*args, **kwargs)
53                 local_cache[key] = (ret,datetime.datetime.now().timestamp())    #
54 
55             return local_cache[key]
56         return wrapper
57     return _cache
58 
59 def logger(fn):
60     @wraps(fn)
61     def wrapper(*args,**kwargs):
62         start = datetime.datetime.now()
63         ret = fn(*args,**kwargs)
64         delta = (datetime.datetime.now() - start).total_seconds()
65         print(delta)
66         return ret
67     return wrapper
68 
69 @logger
70 @cache(5)
71 def add(x,y=5):
72     time.sleep(3)
73     return x+y
74 
75 add(4)
76 add(4,5)
77 add(x=4,y=5)
78 add(4,y=5)
79 add(y=5,x=4)
80 
81 time.sleep(5)
82 print('*'*20)
83 add(4)
84 add(4,5)
85 add(x=4,y=5)
86 add(4,y=5)
87 add(y=5,x=4)
 1 #2 题目:写一个命令分发器
 2 """
 3 1.程序员可以方便的注册函数到某一个命令,用户输入命令时,路由到注册的函数
 4 2.如果此命令没有对应的注册函数,执行默认函数
 5 3.用户输入用input(">>")
 6 """
 7 '''分析
 8 输入命令映射到一个函数,并执行这个函数。应该是cmd_tbl[cmd]=fn的形式,字典正好合适。
 9 如果输入了某一个cmd命令后,没有找到函数,就要调用缺省的函数执行,这正好是字典缺省参数。
10 cmd是字符串
11 
12 实现主要功能后会发现代码布局很丑陋,最好是不要将所有的函数和字典都在全局中定义!
13 如何改进呢?将reg函数封装成装饰器,并用它来注册函数。
14 
15 重复注册的问题
16 如果函数使用同一个函数名注册,就等于覆盖了原来的cmd到fn的关系,这样的逻辑也是合理的。
17 也可以加一个判断,如果key已经存在,重复注册,抛出异常。看业务需求。
18 
19 注销
20 有注册就应该有注销
21 一般来说注销是要有权限的,但是什么样的人拥有注销的权限。
22 '''
23 command_dict = {}
24 
25 #注册(带参装饰器)
26 def dispatch():
27 
28     def req(name):
29         def wrapper(fn):
30             command_dict[name]=fn
31         return wrapper
32 
33     def defaultfunc():
34         print('Unkown command!')
35 
36     def dispatchfunc():
37         while True:
38             userinput = input('>>')
39             if userinput.strip() == 'quit':
40                 return
41             if userinput in command_dict.keys():
42                 command_dict.get(userinput,defaultfunc)()
43             else:
44                 defaultfunc()
45     return req,dispatchfunc
46 
47 req,dispatchfunc = dispatch()
48 #自定义函数
49 @req('cy')    # f1=req('cy')(f1)
50 def f1():
51     print('chengyu')
52 @req('py')
53 def f2():
54     print('python')
55 
56 dispatchfunc()

 装饰器的应用

 

装饰器应用场景

 作业

做一枚奔跑的老少年!
原文地址:https://www.cnblogs.com/xiaoshayu520ly/p/10661610.html