Python装饰器和描述符应用


纸上得来终觉浅,绝知此事要躬行。

Python装饰器和描述符应用


在之前的博客文章中,已经介绍了关于装饰器、property和描述符的基本使用和技巧。其中,装饰器了解了嵌套的装饰器、不带参数的装饰器、带参数的装饰器、给函数用的类装饰器和给类用的函数装饰器。而这里,我们说说关于三者的生产应用场景。

1. lru-cache

下面是不使用缓存的情况下计算斐波拉切数列,利用timeit模块中Timerfib(30)重复计算10次之后,得到的时间为3.4秒。其中,我们可以看出斐波拉切数列的n值越大计算花费的时间就越长,且递归运算中有重复调用的。

# 使用缓存的情况
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)
In [1]: from timeit import Timer

In [2]: t = Timer('fib(30)', 'from func import fib')

In [3]: t.timeit(10)
Out[3]: 3.488710475998232
In [4]: # fib(5) = fib(4) + fib(3)
In [4]: # fib(6) = fib(5) + fib(4)

所以,我们可以以n为单位,缓存结果以备下次继续使用。我们可以使用之前学习到的装饰器来实现这个缓存机制。

from functools import wraps


def memoize(fn):
    cache = dict()

    @wraps(fn)
    def memoized_fn(*args):
        if args in cache:
            return cache[args]
        result = fn(*args)
        cache[args] = result
        return result
    return memoized_fn


@memoize
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)
In [1]: from timeit import Timer

In [2]: t = Timer('fib(30)', 'from func import fib')

In [3]: t.timeit(10)
Out[4]: 4.968998837284744e-06

而在Python3的标准库functools中存在一个高级模块lru-cache,它用于提供缓存。所以不需要我们去重复造这个轮子,而且我们造的也不一定十分好用。

另外,我认为大家除了使用这种标准库中提供的函数和方法,可以翻翻源代码看看它的实现。因为,有可能我们有时候需要自己去造轮子。可以看看标准库中他们是怎么实现的,然后对比一下其好的地方,自己吸收之后在自己的代码中使用。

from functools import lru_cache

@lru_cache(maxsize=None)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)
In [1]: from func import fib

In [2]: fib(10)
Out[2]: 55

In [3]: fib.cache_info()
Out[3]: CacheInfo(hits=8, misses=11, maxsize=None, currsize=11)

2. login_required

这个login_requiredWeb开发里面会经常被看到,也就是当访问某个路由的时候,首先会判断你是否登录。当然,也可以加一些别的装饰器在对应的视图上,比如说数据格式的验证、权限的验证、页面缓存等等的内容。

下面这个示例是Flasklogin里面的login_required的实现,首先判断一下使用的请求方法是否在EXEMPT_METHODS这个变量里面,或者这个_login_disabledTrue,返回的结果就是不验证你的登录状态。否则,就会判断你是否已经通过is_authenticated身份验证。而如果没验证通过的话,就会抛出一个错误,通常都会是404的页面。

from functools import wrapsdef

login_required(func):
    @wraps(func)
    def decorated_view(*args, **kwargs):
        if request.method in EXEMPT_METHODS:
            return func(*args, **kwargs)
        elif current_app.login_manager._login_disabled:
            return func(*args, **kwargs)
        elif not current_user.is_authenticated:
            return current_app.login_manager.unauthorized()
        return func(*args, **kwargs)
    return decorated_view

3. cached_property

Web开发里面还会用到一个cached_property。虽然看起来它和普通的property属性没有什么区别,但是它可以对property属性的计算结果进行缓存,提高效率。

In [1]: class Foo:
   ...:     @cached_property
   ...:     def foo(self):
   ...:         print('calculate something important here')
   ...:         return 42
   ...:

In [2]: foo = Foo()

In [3]: foo.foo
calculate something important here
Out[3]: 42

In [4]: foo.foo
Out[4]: 42

In [5]: foo.foo
Out[5]: 42

我们可以看一下Flask里面对于这个cached_property的实现。可以看到这是一个装饰函数的类装饰器,在valueobj.__dict__这个字典中获取不到对应的值时设置为我们定义的_missing对象,其用于标记这个值并没有被缓存,需要计算之后才能缓存起来。

这里的_missing可不可以使用Nonde或者是0呢?当然是不可以的,应为我们返回的值有可能是Nonde或者0,这样我们就没有办法辨认其是否需要被缓存。

这里的__init__做的工作就和标准库functools中的wraps是一样的,只是wraps是将其所有的属性都传递了,而这里是选择性的继承了几个属性而已。

class _Missing(object):
    def __repr__(self):
        return 'no value'

    def __reduce__(self):
        return'_missing'

_missing = _Missing()


class cached_property(property):
    def __init__(self, func, name=None, doc=None):
        self.__name__ = name or func.__name__
        self.__module__ = func.__module__
        self.__doc__ = doc or func.__doc__
        self.func = func

    def __set__(self, obj, value):
        obj.__dict__[self.__name__] = value

    def __get__(self, obj, type=None):
        if obj is None:
            returnself
        value = obj.__dict__.get(self.__name__, _missing)
        if value is _missing:
            value = self.func(obj)
            obj.__dict__[self.__name__] = value
        return value

4. lazy property

如果我们在初始化的时候,其中_get_all_relatives是一个执行很耗时的方法去拿到这个属性的值,这样当初始化的时候就非常耗费时间。

class Person:
    def __init__(self, name):
        self.name = name
        self.relatives = self._get_all_relatives()

    def _get_all_relatives(self):
        # 这个方法执行很耗时
        ...

所以,这样做是不对的,我们可以进行延迟计算。即在初始化的时候不进行计算,而是在用的时候在计算,从而节省了程序的内存需求和提高了效率。

通常是如下这种用法,首先设置了_relatives变量为一个空列表,在程序调用relatives属性的时候,先判断是否为空,如果为空则计算对应的值,如果不为空则直接抛出_relatives的值。

class Person:
    def __init__(self, name):
        self.name = name
        self._relatives = []

    @property
    def relatives(self):
        if not self._relatives:
            self._relatives = self._get_all_relatives()
        return self._relatives

    def _get_all_relatives(self):
        # 这个方法执行很耗时
        ...

但是,这里有一个问题,那就是如果就是需要返回一个空的列表,那么还是会执行这个耗时的计算的。所以需要我们对其进行特殊化的定制,这样就不会出现重复计算的问题了。

当然,这里也是用使用之前的那个cached_property,而这里使用lazy_property进行延迟计算。这里使用hasattr函数来判断其是否存在一个attr_name的属性,如果不存在在计算,存在直接返回结果。

def lazy_property(fn):
    attr_name = '_lazy_' + fn.__name__

    @property
    def _lazy_property(self):
        if not hasattr(self, attr_name):
            setattr(self, attr_name, fn(self))
        return getattr(self, attr_name)
    return _lazy_property


class Person:
    def __init__(self, name):
        self.name = name

        @lazy_property
        def relatives(self):
            # 这个方法执行很耗时
            ...

文章作者: Escape
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 Escape !
  目录