Python中的functools模块

朴灿烈づ我的快乐病毒、 2022-04-25 07:08 301阅读 0赞

# Python中的functools模块 #

1.  update\_wrapper(wrapper,wrapped,assigned=WRAPPER\_ASSIGNMENTS,updated=WRAPPER\_UPDATES)
    
     *  功能类似上面代码中copy\_property功能，帮助修改包装函数属性
     *  wrapper 包装还是、被更新者（需要更新者）
     *  wrapped 被包装函数、数据源
     *  assigned=WRAPPER\_ASSIGNMENTS 其中WRAPPER\_ASSIGNMENTS是一个元组
        
         *  WRAPPER\_ASSIGNMENTS = (’\_\_module\_\_’, ‘\_\_name\_\_’, ‘\_\_qualname\_\_’, ‘\_\_doc\_\_’,  
            ‘\_\_annotations\_\_’)
     *  updated=WRAPPER\_UPDATES 其中 WRAPPER\_UPDATES是一个元组
        
         *  WRAPPER\_UPDATES = (’\_\_dict\_\_’,)
     *  注意：在update\_wrapper函数调用后，会新**增加一个\_\_wrapped\_\_属性，记录了被包装函数wrapped**
     *  原码如下：  
        ![decorator006][]
     *  使用update\_wrapper函数修改属性值如下：
    
        import datetime,time,functools
        
        #定义修改注解函数，并将修改注解函数柯里化
        def logg(fn):
            def copy_property(tofn):
                tofn.__name__ = fn.__name__
                tofn.__doc__ = fn.__doc__
                return tofn  #注意，这里需要将原函数返回。不然元函数值为none
            return copy_property
        
        def strengthen(fn):
            """我是strengthen方法的注解"""
        # @logg(fn) #等效于 wrapper = logg(fn)(wrapper)
            def wrapper(*args,**kwargs):
                """我是wraper方法的注解"""
                starttime = datetime.datetime.now()
                fn(*args,**kwargs) #这里使用原理的加法器做运算
                elapsedtime = datetime.datetime.now() - starttime
                print("耗时：{}".format(elapsedtime.total_seconds()))
            functools.update_wrapper(wrapper,fn)  #效果类似于logg(fn)(wrapper)
            return wrapper  
        
        @strengthen    #等效于add = strengthen(add)
        def add(x,y):
            """我是add方法的注解"""
            print("开始计算{}+{}".format(x,y))
            time.sleep(1)
            hh = x + y
            print("结果为：{}".format(hh))
            return hh
        
        add(5,4)
        print(add.__name__)
        print()
        add.__wrapped__(5,6)  #__wrapped属性值中保存了原来的被包装函数
        # datetime.datetime.now()
    
     *  运行结果如下：
    
    ![decorator007][]
    
     *  其代码还可以修改为如下方式：
    
        import datetime,time,functools
        
        #定义修改注解函数，并将修改注解函数柯里化
        def logg(fn):
            def copy_property(tofn):
                tofn.__name__ = fn.__name__
                tofn.__doc__ = fn.__doc__
                return tofn  #注意，这里需要将原函数返回。不然元函数值为none
            return copy_property
        
        def strengthen(fn):
            """我是strengthen方法的注解"""
        # @logg(fn) #等效于 wrapper = logg(fn)(wrapper)
            @functools.wraps(fn) #与logg(fn)类似
            def wrapper(*args,**kwargs):
                """我是wraper方法的注解"""
                starttime = datetime.datetime.now()
                fn(*args,**kwargs) #这里使用原理的加法器做运算
                elapsedtime = datetime.datetime.now() - starttime
                print("耗时：{}".format(elapsedtime.total_seconds()))
        # functools.update_wrapper(wrapper,fn) #效果类似于logg(fn)(wrapper)
            return wrapper 
        
        @strengthen    #等效于add = strengthen(add)
        def add(x,y):
            """我是add方法的注解"""
            print("开始计算{}+{}".format(x,y))
            time.sleep(1)
            hh = x + y
            print("结果为：{}".format(hh))
            return hh
        
        add(5,4)
        print(add.__name__)
        print()
        add.__wrapped__(5,6)  #__wrapped属性值中保存了原来的被包装函数
        # datetime.datetime.now()
    
     *  运行结果如下：  
        ![decorator008][]
    
    上面改造中使用了@functools.wraps(fn)
2.  wraps(wrapped,assigned = WRAPPER\_ASSIGNMENTS,updated = WRAPPER\_UPDATES)
    
     *  类似logg功能
     *  wrapped被包装函数
     *  assigned=WRAPPER\_ASSIGNMENTS 其中WRAPPER\_ASSIGNMENTS是一个元组
        
         *  WRAPPER\_ASSIGNMENTS = (’\_\_module\_\_’, ‘\_\_name\_\_’, ‘\_\_qualname\_\_’, ‘\_\_doc\_\_’,  
            ‘\_\_annotations\_\_’)
     *  updated=WRAPPER\_UPDATES 其中 WRAPPER\_UPDATES是一个元组
        
         *  WRAPPER\_UPDATES = (’\_\_dict\_\_’,)
     *  注意：会新**增加一个\_\_wrapped\_\_属性，记录了被包装函数wrapped**
     *  wraps函数原码：  
        ![decorator009][]
3.  reduce(function,iterable\[,initializer\]) \#归纳计算。 通过指定函数将集合中的数据按照指定方式规整为一个数据。
    
     *  function 是个函数，用来定义规整方式
     *  iterables 可迭代对象
     *  initializer 设置规整时的初始值，默认值为None，会获取iterables中的第一个元素作为初始值。
     *  Python官方伪代码如下：
    
        def reduce(function, iterable, initializer=None):
            it = iter(iterable)
            if initializer is None:
                value = next(it)
            else:
                value = initializer
            for element in it:
                value = function(value, element)
            return value
4.  partial(func, \*args, \*\*keywords)->fun \#偏函数。
    
     *  偏函数，把函数部分的参数固定下来，相当于为部分的参数添加了一个固定的默认值，形成一个新的函数并返回
     *  从partial生成的新函数，是对函数的封装
     *  func \#需要封装的函数
     *  \*args \#需要固定的位置参数
     *  \*\*kwywords \#需要固定的关键字参数
     *  简单原码如下：
    
        def partial(func, *args, **keywords):
            def newfunc(*fargs, **fkeywords):
                newkeywords = keywords.copy()
                newkeywords.update(fkeywords)
                return func(*args, *fargs, **newkeywords)
            newfunc.func = func
            newfunc.args = args
            newfunc.keywords = keywords
            return newfunc
    
     *  示例：
    
        >>> from functools import partial
        >>> basetwo = partial(int, base=2)
        >>> basetwo.__doc__ = 'Convert base 2 string to an int.'
        >>> basetwo('10010')
        18

## lru\_cache缓存装饰器 ##

*  functools.lru\_cache(maxsize=128,typed=False) \#cache缓存。为函数添加缓存机制。通过一个字典缓存被装饰函数的调用和返回值。
    
    1.  Least-recently-used装饰器。lru,最近最少使用。cache缓存
    2.  maxsize:如果maxsize设置为None,则禁用LRU功能，并且缓存可以无限制增长。当maxsize是二的幂时，LRU功能执行得最好
    3.  如果typed设置为True,则不同类型的函数参数将单独缓存。例如f(3)和f(3.0)将被视为具有不同结果的不同调用
    4.  示例
    
        import functools
        import time
        
        @functools.lru_cache()
        def add(x,y):
            time.sleep(1)
            return x + y
        
        print(add(4,5))
        print(add(4.0,5))
        print(add(4,6))
        print("--------------")
        print(add(4,5)) #会使用缓存，无需等待，直接出结果
        print(add(4.0,5))  #会使用缓存，无序等待，直接出结果
        print(add(4,6)) #会使用缓存，无序等待，直接出结果
        print("-------------------")
        print(add(4,y=6)) #无法使用缓存
        print(add(x=4,y=6))  #无法使用缓存
        print(add(y=6,x=4)) #无法使用缓存
    
    ![funct\_001][funct_001]
 *  lru\_cache装饰器
    
    1.  斐波拉契数列递归方法的改造
    
        import functools,datetime
        
        def fib(n):
            return 1 if n<3 else fib(n-1)+fib(n-2)
        
        @functools.lru_cache(maxsize=None)
        def fib2(n):
            return 1 if n<3 else fib2(n-1)+fib2(n-2)
        
        dt = datetime.datetime.now()
        print([fib2(i+1) for i in range(35)])
        dt1 = datetime.datetime.now()
        print("（fib2）使用lru_cache缓存技术,用时：{}".format((dt1-dt).total_seconds()))
        print([fib(i+1) for i in range(35)])
        dt2 = datetime.datetime.now()
        print("(fib)不使用lru_cache缓存技术,用时：{}".format((dt2-dt1).total_seconds()))
    
    ![funct\_002][funct_002]
 *  lru\_cache装饰器应用
    
    1.  使用前提
        
         *  同样的函数参数一定得到同样的结果
         *  函数执行时间很长，且要多次执行
    2.  本质是函数调用函数=》返回值
    3.  缺点
        
         *  不支持缓存过期，key无法过期、失效
         *  不支持清除操作
         *  不支持分布式，是一个单机的缓存
    4.  适用场景，单机上需要空间换时间的地方，可以用缓存来将计算变成快速的查询
 *  实现一个cache装饰器，实现可过期被清除的功能
    
    1.  简化设计，函数的形参定义不包含可变位置参数、可变关键字参数和keyword-only参数 可以不考虑缓存大小，也不用考虑缓存满了之后的换出问题

import inspect,functools,datetime
    
    def xdd_time(fn):
        """ 时间统计，统计函数执行时间 :param fn: 需要装饰的函数 :return: 包装后的fn """
        @functools.wraps(fn)
        def wrapper(*args,**kwargs):
            start = datetime.datetime.now()
            req = fn(*args,**kwargs)
            timeout = datetime.datetime.now() - start
            print("{}方法耗时：{}，args={},kwargs={},\n执行结果：req={}".format(fn.__name__,timeout.total_seconds(),args,kwargs,req))
            return req
        return wrapper
    
    def xdd_cache(duration = 5):
        def _xdd_cache(fn):
            """ 为函数生成缓存机制，具有有效时长功能 :param fn: :param duration:缓存时长，单位秒，如果为None表示全部缓存，不计算时长 :return: """
            xdd_dict = { } #定义一个字典，用来存参数与对应的计算结果
            sig = inspect.signature(fn) #对函数进行签名
    
            @functools.wraps(fn)
            def wrapper(*args,**kwargs):
                paramestr = getparamestr(args,kwargs) #根据参数获取字符串
                req = xdd_dict.get(paramestr,None)
                if req is None:
                    req = fn(*args,**kwargs),datetime.datetime.now()
                    xdd_dict[paramestr] = req
                elif not duration is None and (datetime.datetime.now() - req[1]).total_seconds() > duration: #超时
                    print("缓存超时")
                    req = fn(*args, **kwargs),datetime.datetime.now()
                    xdd_dict[paramestr] = req
                return req[0]
    
            def getparamestr(args,kwargs)->str:
                """ 格式化参数，将参数格式化为指定格式的字符串 :param args: :param kwargs: :return: """
                parames = sig.parameters
                pstr = ()
                isargs = True #标记是否还有位置参数
                kwdict = { }
                for k,(name,v) in enumerate(parames.items()):
                    v:inspect.Parameter = v
                    if name == "args": #位置参数已经全部匹配完成
                        isargs = False
                        pstr += ("args",tuple(args[k+1:])) #记录剩下的位置参数
                        continue
                    if name == "kwargs": #关键字位置参数已经匹配完
                        pstr += ("kwargs",tuple((k,kwargs[k]) for k in kwargs.keys()-kwdict.keys()))
                        break
                    if isargs:  # 位置参数还没匹配完
                        if k<len(args):
                            pstr += (name, args[k])
                        else: #实际传参中位置参数已经传完
                            kwval = kwargs.get(name,v.default)
                            kwdict[name] = kwval  # 记录已经出现的位置参数
                            pstr += (name,kwval)
                    else:
                        kwval = kwargs.get(name, v.default)
                        kwdict[name] = kwval #记录已经出现的位置参数
                        pstr += (name,kwval)
                    # print(k,name,v.name,v.annotation,v.default)
                return pstr
            return wrapper
        return _xdd_cache
    
    # @gdy_cache
    # def add(x,y:int=5,*args,z=4,b:int,c=6,**kwargs):
    # import time
    # time.sleep(3)
    # return x+y
    @xdd_time
    @xdd_cache()
    def add(x=4,y=5):
        import time
        time.sleep(3)
        return x+y
    print("开始计算")
    print(1,add(4,5))
    import time
    time.sleep(5)
    print(2,add(4))
    # print(3,add(y=5))
    # print(4,add(x=4,y=5))
    # print(5,add(y=5,x=4))
    # print(6,add())

![homework190422\_003][homework190422_003]

[decorator006]: /images/20220126/ffc38384a9ed4c39aa44b5676d2cee61.png
[decorator007]: /images/20220126/49fbefbc29f6475fa78d9ddbd0c07573.png
[decorator008]: /images/20220126/4bc80773a8674ed0aca2efd526b7a6d0.png
[decorator009]: /images/20220126/db3e24e397714767ac47bd5a407e0906.png
[funct_001]: /images/20220126/53cc82d9eb5f4567a4177b6ae0982ebe.png
[funct_002]: /images/20220126/244f0a27a9be47d79f8ab2059a7333b5.png
[homework190422_003]: /images/20220126/2c6b7ffc6675465cb8ce8a28a285719a.png