内置标准库之functools

**Python**语言自带的 functools 模块提供了一些常用的高阶函数，使用了这些函数可以大大简化我们的代码。如可以固定函数参数的partial、封装函数属性的wraps、保存计算缓存数据的lru_cache等等。

---

• 官网地址

  • functools — 官网文档
  • functools — 代码地址

• 函数概览

编号	模块名称	解释说明
1	functools.reduce()	对可迭代对象进行合并计算
2	functools.cmp_to_key()	将旧式的比较函数转换为关键字函数
3	functools.total_ordering()	根据已有的函数自动补全比较运算符函数
4	functools.partial()	偏函数
5	functools.partialmethod()	类似于偏函数 partial 且仅作用于方法
6	functools.update_wrapper()	封装函数的属性
7	functools.wraps()	封装属性，就是用 partial 对 update_wrapper 做了包装
8	functools.lru_cache()	保存计算缓存数据
9	functools.singledispatch()	将函数转换为单调度通用函数
10	functools.singledispatchmethod()	将函数转换为单调度通用函数
11	functools.cached_property()	用于在实例生命周期内缓存已计算特征属性

---

1. reduce

对可迭代对象进行合并计算

语法规则

• functools.reduce(function, iterable[, initializer])

详细解释

• 该函数与Python内置的reduce()函数相同，主要用于编写兼容Python3的代码。

背景说明

• 这个 functools.reduce 就是 Python 2 内建库中的 reduce ，它之所以出现在这里就是因为 Guido 的独裁，他并不喜欢函数式编程中 “map-reduce” 的思想概念。因此打算将 map 和 reduce 两个函数移出内建函数库，最后在社区的强烈反对中将map函数保留在了内建库中。但是 Python3 内建的 map 函数返回的是一个迭代器对象，而 Python2 中会 eagerly 生成一个 list ，这点是个坑，所以以后使用时需要多加注意。

实例演示

• 函数基本使用方式

$ reduce(lambda x, y: x+y, [1, 2, 3, 4, 5])
((((1+2)+3)+4)+5)

• 函数大致实现方式

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


$ reduce(lambda x, y: x+y, [1, 2, 3, 4, 5])
15

当没有初始化程序时，具有单个项目的序列会自动减少到该值。除非提供了初始化程序，否则空列表会产生错误。

import functools

def do_reduce(a, b):
    print('do_reduce({}, {})'.format(a, b))
    return a + b

print('Single item in sequence:',
      functools.reduce(do_reduce, [1]))

print('Single item in sequence with initializer:',
      functools.reduce(do_reduce, [1], 99))

print('Empty sequence with initializer:',
      functools.reduce(do_reduce, [], 99))

try:
    print('Empty sequence:', functools.reduce(do_reduce, []))
except TypeError as err:
    print('ERROR: {}'.format(err))

---

2. cmp_to_key

将旧式的比较函数转换为关键字函数

语法规则

• functools.cmp_to_key(func)

详细解释

• 该函数用于将旧式的**比较函数转换为关键字函数**。
• 在 Python3 中，有很多地方都不再支持旧式的比较函数，即不再支持参数比较的方式指定比较方式了，所以此时可以使用 cmp_to_key() 进行转换。

背景说明

• 旧式的比较函数
  • 接收两个参数，返回比较的结果
  • 返回值小于零则前者小于后者
  • 返回值大于零则前者大于后者
  • 返回值等于零则两者相等
• 关键字函数
  • 接收一个参数，返回其对应的可比较对象
  • 例如 sorted()、 min()、 max()、 heapq.nlargest()、 heapq.nsmallest()、 itertools.groupby() 等都可作为关键字函数

实例演示

• 函数基本使用方式

# 正序
$ sorted(range(5), key=cmp_to_key(lambda x, y: x-y))
[0, 1, 2, 3, 4]

# 逆序
$ sorted(range(5), key=cmp_to_key(lambda x, y: y-x))
[4, 3, 2, 1, 0]

• 函数大致实现方式

def cmp_to_key(mycmp):
    """Convert a cmp= function into a key= function"""
    class K(object):
        __slots__ = ['obj']
        def __init__(self, obj):
            self.obj = obj
        def __lt__(self, other):
            return mycmp(self.obj, other.obj) < 0
        def __gt__(self, other):
            return mycmp(self.obj, other.obj) > 0
        def __eq__(self, other):
            return mycmp(self.obj, other.obj) == 0
        def __le__(self, other):
            return mycmp(self.obj, other.obj) <= 0
        def __ge__(self, other):
            return mycmp(self.obj, other.obj) >= 0
        __hash__ = None
    return K

try:
    from _functools import cmp_to_key
except ImportError:
    pass

---

3. total_ordering

根据已有的函数自动补全比较运算符函数

语法规则

• functools.total_ordering(cls)

详细解释

• 这是一个类装饰器，用于自动实现类的比较运算，用于简化比较函数的写法。
• 我们只需要在类中实现 __eq__() 方法和以下方法中的任意一个 __lt__()、 __le__()、 __gt__()、 __ge__()，那么 total_ordering() 就能自动帮我们实现余下的几种比较运算。

实例演示

• 基本使用方式

@total_ordering
class Student:
    def __init__(self, lastname, firstname):
        self.lastname = lastname
        self.firstname = firstname

    def __eq__(self, other):
        return ((self.lastname.lower(), self.firstname.lower()) ==
                (other.lastname.lower(), other.firstname.lower()))

    def __lt__(self, other):
        return ((self.lastname.lower(), self.firstname.lower()) <
                (other.lastname.lower(), other.firstname.lower()))


$ dir(Student)
['__doc__', '__eq__', '__ge__', '__gt__', '__le__', '__lt__', '__module__']

• 函数大致实现方式

def total_ordering(cls):
    roots = {op for op in _convert \
             if getattr(cls, op, None) is not getattr(object, op, None)}
    if not roots:
        raise ValueError('must define at least one ordering operation: < > <= >=')
    root = max(roots)
    for opname, opfunc in _convert[root]:
        if opname not in roots:
            opfunc.__name__ = opname
            setattr(cls, opname, opfunc)
    return cls

---

4. partial

调用 partial()函数需要传入三个只读属性，分别是 func、 args 和 keywords。

语法规则

• functools.partial(func[, args][, *keywords])

详细解释

• 函数式编程中有个很重要的概念叫做柯里化，partial偏函数就是实现的一种方式。
• partial() 函数主要用于冻结函数的部分参数，返回一个参数更少、使用更简单的函数对象。

对象属性

• (1) partial.func
  • 可调用或函数，调用partial对象时，会结合新的参数和关键字最终调用func。
• (2) partial.args
  • 默认为最左的位置参数，这些参数会被自动添加到所有调用partial对象时传入的参数前。
  • 也就是说在调用partial对象时不用传入这些参数即可，他们视为恒定的，从而自动添加到func的调用中。
• (3) partial.keywords
  • 当调用partial对象时提供的关键字参数。
  • partial对象在一些方面类似于函数对象：如可调用，弱引用，可以拥有属性。
  • 但是也有一些关键的区别：1.partial对象的 __name__ 和 __doc__ 属性不会自动创建；2. 在类里定义的partial对象使用时更像静态方法，在实例的属性查询过程中不会转变成绑定方法（bound methods, 通过类的实例对象进行属性引用）。

实例演示

• 基本使用方式

>>> from functools import partial

>>> basetwo = partial(int, base=2)
>>> basetwo.__doc__ = 'Convert base 2 string to an int.'
>>> basetwo('10010')
18

• 函数大致实现方式

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

• 复杂一点的示例

import functools

def myfunc(a, b=2):
    print('  called myfunc with:', (a, b))

def show_details(name, f, is_partial=False):
    print('{}:'.format(name))
    print('  object:', f)
    if not is_partial:
        print('  __name__:', f.__name__)
    if is_partial:
        print('  func:', f.func)
        print('  args:', f.args)
        print('  keywords:', f.keywords)
    return

myfunc('a', 3)
p1 = functools.partial(myfunc, b=4)
p2 = functools.partial(myfunc, 'default a', b=99)

---

5. partialmethod

类似于偏函数 partial 且仅作用于方法

语法规则

• functools.partialmethod(func, *args, **keywords)

详细解释

• partialmethod 是 Python 3.4 中新引入的装饰器，作用基本类似于偏函数 partial， 不过 partialmethod 仅作用于方法。

实例演示

• 基本使用方式

>>> class Cell(object):
...     def __init__(self):
...         self._alive = False
...     @property
...     def alive(self):
...         return self._alive
...     def set_state(self, state):
...         self._alive = bool(state)
...
...     set_alive = partialmethod(set_state, True)
...     set_dead = partialmethod(set_state, False)
...

>>> c = Cell()
>>> c.alive
False

>>> c.set_alive()
>>> c.alive
True

• 函数大致实现方式

class partialmethod(object):
    def __init__(self, func, *args, **keywords):
        if not callable(func) and not hasattr(func, "__get__"):
            raise TypeError("{!r} is not callable or a descriptor".format(func))

        if isinstance(func, partialmethod):
            self.func = func.func
            self.args = func.args + args
            self.keywords = func.keywords.copy()
            self.keywords.update(keywords)
        else:
            self.func = func
            self.args = args
            self.keywords = keywords

    def __repr__(self):
        args = ", ".join(map(repr, self.args))
        keywords = ", ".join("{}={!r}".format(k, v)
                                 for k, v in self.keywords.items())
        format_string = "{module}.{cls}({func}, {args}, {keywords})"
        return format_string.format(module=self.__class__.__module__,
                                    cls=self.__class__.__qualname__,
                                    func=self.func,
                                    args=args,
                                    keywords=keywords)

    def _make_unbound_method(self):
        def _method(*args, **keywords):
            call_keywords = self.keywords.copy()
            call_keywords.update(keywords)
            cls_or_self, *rest = args
            call_args = (cls_or_self,) + self.args + tuple(rest)
            return self.func(*call_args, **call_keywords)
        _method.__isabstractmethod__ = self.__isabstractmethod__
        _method._partialmethod = self
        return _method

    def __get__(self, obj, cls):
        get = getattr(self.func, "__get__", None)
        result = None
        if get is not None:
            new_func = get(obj, cls)
            if new_func is not self.func:
                result = partial(new_func, *self.args, **self.keywords)
                try:
                    result.__self__ = new_func.__self__
                except AttributeError:
                    pass
        if result is None:
            result = self._make_unbound_method().__get__(obj, cls)
        return result

    @property
    def __isabstractmethod__(self):
        return getattr(self.func, "__isabstractmethod__", False)

• 复杂一点的示例

import functools

def standalone(self, a=1, b=2):
    print('  called standalone with:', (self, a, b))
    if self is not None:
        print('  self.attr =', self.attr)

class MyClass:
    def __init__(self):
        self.attr = 'instance attribute'

    method1 = functools.partialmethod(standalone)
    method2 = functools.partial(standalone)

>>> o = MyClass()

>>> print('standalone')
>>> standalone(None)
>>> print()
standalone
  called standalone with: (None, 1, 2)

>>> print('method1 as partialmethod')
>>> o.method1()
>>> print()
method1 as partialmethod
  called standalone with: (<__main__.MyClass object at
0x1007b1d30>, 1, 2)
  self.attr = instance attribute

>>> print('method2 as partial')
>>> o.method2()
method2 as partial
ERROR: standalone() missing 1 required positional argument:
'self'

---

6. update_wrapper

封装函数的属性

语法规则

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

详细解释

• 该函数用于更新包装函数 wrapper ，使它看起来像原函数一样。
• update_wrapper 的功能，它可以把被封装函数的 __name__、__module__、__doc__ 和 __dict__ 都复制到封装函数去

参数说明

• 可选的参数是一个元组，assigned 元组指定要直接使用原函数的值进行替换的属性，updated 元组指定要对照原函数进行更新的属性。
• 这两个参数的默认值分别是模块级别的常量：WRAPPER_ASSIGNMENTS 和 WRAPPER_UPDATES。
• 前者指定了对包装函数的 __name__, __module__, __doc__ 属性进行直接赋值，而后者指定了对包装函数的 __dict__ 属性进行更新。
• 如果没有对包装函数进行更新，那么被装饰后的函数所具有的元信息就会变为包装函数的元信息，而不是原函数的元信息。

重点说明

• wraps 其实是 update_wrapper 的特殊化，实际上 wraps(wrapped) 相当于 partial(update_wrapper, wrapped=wrapped, **kwargs)。

实例演示

WRAPPER_ASSIGNMENTS = ('__module__', '__name__', '__doc__')
WRAPPER_UPDATES = ('__dict__',)

def update_wrapper(wrapper, wrapped,
                   assigned=WRAPPER_ASSIGNMENTS,
                   updated=WRAPPER_UPDATES):
    for attr in assigned:
        setattr(wrapper, attr, getattr(wrapped, attr))
    for attr in updated:
        getattr(wrapper, attr).update(getattr(wrapped, attr, {}))
    return wrapper

---

7. wraps

封装属性，就是用 partial 对 update_wrapper 做了包装

语法规则

functools.wraps(wrapped,
                assigned=WRAPPER_ASSIGNMENTS,
                updated=WRAPPER_UPDATES)

详细解释

• functools.wraps 就是用 partial 对 update_wrapper 做了固定参数的封装。

实例演示

• 基本使用方式

>>> from functools import wraps

>>> def my_decorator(f):
...     @wraps(f)
...     def wrapper(*args, **kwds):
...         print('Calling decorated function')
...         return f(*args, **kwds)
...     return wrapper
...

>>> @my_decorator
... def example():
...     """Docstring"""
...     print('Called example function')
...

>>> example()
Calling decorated function
Called example function

>>> example.__name__
'example'

>>> example.__doc__
'Docstring'

• 函数大致实现方式

def wraps(wrapped,
          assigned=WRAPPER_ASSIGNMENTS,
          updated=WRAPPER_UPDATES):
    return partial(update_wrapper,
                   wrapped=wrapped,
                   assigned=assigned,
                   updated=updated)

---

8. lru_cache

保存计算缓存数据

语法规则

• functools.lru_cache(maxsize=128, typed=False)

详细解释

• 装饰器用一个记忆可调用来包装一个函数，可以保存最大的缓存次数。当使用相同的参数调用时，直接从缓存中取值，这样可以节省计算带来的时间消耗。
• 由于字典用于缓存结果，因此函数的位置和关键字参数必须是可哈希的。
• 当 maxsize 参数设置为 None 时，LRU 功能被禁用，缓存可以无限制地增长。
• 如果 typed 设置为 true，则不同类型的函数参数将被单独缓存。例如，f(3) 和 f(3.0) 将被视为具有不同结果的不同调用，进行两次缓存。
• 高速缓存的大小限制确保缓存不会在长时间运行的进程(如Web服务器)上增长。

实例演示

• 基本使用方式

# 用于静态Web内容的LRU缓存的示例
@lru_cache(maxsize=32)
def get_pep(num):
    'Retrieve text of a Python Enhancement Proposal'
    resource = 'http://www.python.org/dev/peps/pep-%04d/' % num
    try:
        with urllib.request.urlopen(resource) as s:
            return s.read()
    except urllib.error.HTTPError:
        return 'Not Found'


>>> for n in 8, 290, 308, 320, 8, 218, 320, 279, 289, 320, 9991:
...     pep = get_pep(n)
...     print(n, len(pep))

>>> get_pep.cache_info()
CacheInfo(hits=3, misses=8, maxsize=32, currsize=8)

# 实现带缓存的斐波拉切数列

@lru_cache(maxsize=None)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)


>>> [fib(n) for n in range(16)]
[0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610]

>>> fib.cache_info()
CacheInfo(hits=28, misses=16, maxsize=None, currsize=16)

• 函数大致实现方式

def lru_cache(maxsize=128, typed=False):
    if maxsize is not None and not isinstance(maxsize, int):
        raise TypeError('Expected maxsize to be an integer or None')

    def decorating_function(user_function):
        wrapper = _lru_cache_wrapper(user_function, maxsize, typed, _CacheInfo)
        return update_wrapper(wrapper, user_function)

    return decorating_function

---

9. singledispatch

将函数转换为单调度通用函数

语法规则

• functools.singledispatch(default)

详细解释

• 将函数转换为单调度通用函数。
• register() 函数定义参数类型，可叠加使用。
• 要检查通用函数为给定类型选择哪个实现，请使用 dispatch() 属性。

实例演示

>>> from functools import singledispatch

# 定义fun函数
>>> @singledispatch
... def fun(arg, verbose=False):
...     if verbose:
...         print("Let me just say,", end=" ")
...     print(arg)

# arg为int
>>> @fun.register(int)
... def _(arg, verbose=False):
...     if verbose:
...         print("Strength in numbers, eh?", end=" ")
...     print(arg)

# arg为list
>>> @fun.register(list)
... def _(arg, verbose=False):
...     if verbose:
...         print("Enumerate this:")
...     for i, elem in enumerate(arg):
...         print(i, elem)

# 可叠加
>>> @fun.register(float)
... @fun.register(Decimal)
... def fun_num(arg, verbose=False):
...     if verbose:
...         print("Half of your number:", end=" ")
...     print(arg / 2)
...

>>> fun("Hello, world.")
Hello, world.

>>> fun("test.", verbose=True)
Let me just say, test.

>>> fun(42, verbose=True)
Strength in numbers, eh? 42

>>> fun(['spam', 'spam', 'eggs', 'spam'], verbose=True)
Enumerate this:
0 spam
1 spam
2 eggs
3 spam

>>> fun_num is fun
False

>>> fun(1.23)
0.615

# 要检查通用函数为给定类型选择哪个实现，请使用dispatch()属性
>>> fun.dispatch(float)
<function fun_num at 0x1035a2840>

>>> fun.dispatch(dict)    # note: default implementation
<function fun at 0x103fe0000>

# 要访问所有注册的实现，请使用只读注册表属性
>>> fun.registry.keys()
dict_keys([<class 'NoneType'>, <class 'int'>, <class 'object'>,
          <class 'decimal.Decimal'>, <class 'list'>,
          <class 'float'>])

>>> fun.registry[float]
<function fun_num at 0x1035a2840>

>>> fun.registry[object]
<function fun at 0x103fe0000>

---

10. singledispatchmethod

将函数转换为单调度通用函数

添加了新的 functools.singledispatchmethod() 可以将普通函数转换为泛型函数的装饰器。泛型函数是指由多个函数组成的函数，可以对不同类型实现相同的操作，调用时应该使用哪个实现由分派算法决定。而 Single dispatch 则是一种泛型函数分派形式，基于单个参数的类型来决定。

from functools import singledispatchmethod
from contextlib import suppress

class TaskManager:
    def __init__(self, tasks):
        self.tasks = list(tasks)

    @singledispatchmethod
    def discard(self, value):
        with suppress(ValueError):
            self.tasks.remove(value)

    @discard.register(list)
    def _(self, tasks):
        targets = set(tasks)
        self.tasks = [x for x in self.tasks if x not in targets]

上面是官方文档中给出的示例代码，想必不太好理解吧，那我们一起看一个简单的示例！下面示例代码中，就是通过 json 序列化一个字段数据，但是因为 now 字段是 datetime() 类型不能直接 json 序列化。

In [1]: from datetime import datetime, date

In [2]: now = datetime.now()

In [3]: d = {'now': now, 'name': 'XiaoMing'}

In [4]: import json

In [5]: json.dumps(d)
----------------------------------------------------------------------
TypeError: Object of type datetime is not JSON serializable

而一般常用的解决方案就是，指定 default 参数，使用函数对其进行格式转化。如果一个对象的 obj 是 datetime 和 date 类型的话，在序列化的时候就直接用 obj.isoformat() 将其转成了字符串，之后在进行显示。

In [7]: def json_encoder(obj):
   ...:     if isinstance(obj, (date, datetime)):
   ...:         return obj.isoformat()
   ...:     raise TypeError(f'{repr(obj)} is not JSON serializable')
   ...:

In [8]: json.dumps(d, default=json_encoder)
Out[8]: '{"now": "2020-10-16T15:42:21.172975", "name": "XiaoMing"}'

如果用 singledispatch 的话，就需要这么写。虽然可以完成任务，但是整体逻辑分散增加了代码行数以及该模式会不必要的增加代码阅读的难度，所以很少在正式环境上面使用到。

In [12]: from functools import singledispatch

In [13]: @singledispatch
    ...: def json_encoder(obj):
    ...:     raise TypeError(f'{repr(obj)} is not JSON serializable')
    ...:

In [14]: @json_encoder.register(date)
    ...: @json_encoder.register(datetime)
    ...: def encode_date_time(obj):
    ...:     return obj.isoformat()
    ...:

In [15]: json.dumps(d, default=json_encoder)
Out[15]: '{"now": "2020-10-16T15:42:21.172975", "name": "XiaoMing"}'

需要知道的是 singledispatch 主要针对的是函数，而对于方法则需要使用新版本引入的 singledispatchmethod 来做。

# 把singledispatch用在方法上就失效了
In [1]: from functools import singledispatch

In [2]: class Dispatch:
   ...:     @singledispatch
   ...:     def foo(self, a):
   ...:         return a
   ...:     @foo.register(int)
   ...:     def _(self, a):
   ...:         return 'int'
   ...:     @foo.register(str)
   ...:     def _(self, a):
   ...:         return 'str'
   ...:

In [3]: cls = Dispatch()

# 没有返回 'int'
In [4]: cls.foo(1)
Out[4]: 1

# 没有返回 'str'
In [5]: cls.foo('s')
Out[5]: 's'

# 使用singledispatchmethod生效了
In [6]: from functools import singledispatchmethod

In [7]: class Dispatch:
   ...:     @singledispatchmethod
   ...:     def foo(self, a):
   ...:         return a
   ...:     @foo.register(int)
   ...:     def _(self, a):
   ...:         return 'int'
   ...:     @foo.register(str)
   ...:     def _(self, a):
   ...:         return 'str'
   ...:

In [8]: cls = Dispatch()

In [9]: cls.foo(1)
Out[9]: 'int'

In [10]: cls.foo('s')
Out[10]: 'str'

这种模式还可以用在 classmethod、staticmethod、abstractmethod 等装饰器上，如官网的例子。

class Negator:
    @singledispatchmethod
    @classmethod
    def neg(cls, arg):
        raise NotImplementedError("Cannot negate a")

    @neg.register
    @classmethod
    def _(cls, arg: int):
        return -arg

    @neg.register
    @classmethod
    def _(cls, arg: bool):
        return not arg

---

11. cached_property

用于在实例生命周期内缓存已计算特征属性

添加了新的 functools.cached_property() 装饰器，用于在实例生命周期内缓存已计算特征属性。这是一个非常常用的功能，很多知名 Python 项目都自己实现过它，现在终于进入版本库了。我们这就简单看看其的使用方法和特性。

下面的实例中，我们创建 Dataset 的实例之后调用 d.variance 时，可以看到执行了 variance 方法的逻辑(可以看到对应输出信息)，而再次执行的时候就直接返回缓存的属性的值了。而且标准库中的版本还有一种的特点，就是加了线程锁，防止多个线程一起修改缓存。可以确保在项目中使用多线程并发访问时，不会出现数值的修改，更加安全可靠。需要注意的是其不支持异步调用，如果需要使用在异步中，可以使用 pydanny 提供的 cached-property 这个包。

import functools
import statistics

class Dataset:
   def __init__(self, sequence_of_numbers):
      self.data = sequence_of_numbers

   @functools.cached_property
   def variance(self):
      print('get variance data')
      return statistics.variance(self.data)

>>> from functools_test import Dataset
>>> d = Dataset(40)

>>> d.variance
get variance data
40

>>> d.variance
40