Python子类方法从超类方法继承装饰器

Question

我有一个超类，它有一个retrieve() 方法，它的子类都实现了自己的retrieve() 方法。我希望每个retrieve() 方法都被修饰以在它接收到相同的参数时缓存返回值，而不必在每个子类中修饰该方法。

装饰器似乎不是被继承的。我可能会调用超类的方法来设置缓存，但目前我的超类引发了一个 NotImplemented 异常，我喜欢这个。

import json
import operator
from cachetools import cachedmethod, TTLCache

def simple_decorator(func):
    def wrapper(*args, **kwargs):
        #check cache
        print("simple decorator")
        func(*args, **kwargs)
        #set cache
    return wrapper


class AbstractInput(object):
    def __init__(self, cacheparams = {'maxsize': 10, 'ttl': 300}):
        self.cache = TTLCache(**cacheparams)
        super().__init__()

    @simple_decorator
    def retrieve(self, params):
        print("AbstractInput retrieve")
        raise NotImplementedError("AbstractInput inheritors must implement retrieve() method")

class JsonInput(AbstractInput):
    def retrieve(self, params):
        print("JsonInput retrieve")
        return json.dumps(params)

class SillyJsonInput(JsonInput):
    def retrieve(self, params):
        print("SillyJsonInput retrieve")
        params["silly"] = True
        return json.dumps(params)

实际结果：

>>> ai.retrieve(params)
ai.retrieve(params)
simple decorator
AbstractInput retrieve
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<string>", line 8, in wrapper
  File "<string>", line 22, in retrieve
NotImplementedError: AbstractInput inheritors must implement retrieve() method
>>> ji.retrieve(params)
ji.retrieve(params)
JsonInput retrieve
'{"happy": "go lucky", "angry": "as a wasp"}'

期望的结果：

>>> ai.retrieve(params)
ai.retrieve(params)
simple decorator
AbstractInput retrieve
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<string>", line 8, in wrapper
  File "<string>", line 22, in retrieve
NotImplementedError: AbstractInput inheritors must implement retrieve() method
>>> ji.retrieve(params)
simple decorator
ji.retrieve(params)
JsonInput retrieve
'{"happy": "go lucky", "angry": "as a wasp"}'

Answer 1

好吧，我似乎可以在超类中“装饰”一个方法，并让子类也继承该装饰，即使该方法在子类中被覆盖，使用元类。在这种情况下，我使用名为 CacheRetrieval 的元类，使用 simple_decorator 装饰 AbstractInput 及其子类中的所有“检索”方法。

def simple_decorator(func):
    def wrapper(*args, **kwargs):
        print("check cache")
        rt = func(*args, **kwargs)
        print("set cache")
        return rt
    return wrapper

class CacheRetrieval(type):
    def __new__(cls, name, bases, attr):
        # Replace each function with
        # a print statement of the function name
        # followed by running the computation with the provided args and returning the computation result
        attr["retrieve"] = simple_decorator(attr["retrieve"])

        return super(CacheRetrieval, cls).__new__(cls, name, bases, attr)


class AbstractInput(object, metaclass= CacheRetrieval):
    def __init__(self, cacheparams = {'maxsize': 10, 'ttl': 300}):
        self.cache = TTLCache(**cacheparams)
        super().__init__()

    def retrieve(self, params):
        print("AbstractInput retrieve")
        raise NotImplementedError("DataInput must implement retrieve() method")


class JsonInput(AbstractInput):
    def retrieve(self, params):
        print("JsonInput retrieve")
        return json.dumps(params)


class SillyJsonInput(JsonInput):
    def retrieve(self, params):
        print("SillyJsonInput retrieve")
        params["silly"] = True
        return json.dumps(params)

这个页面帮助了我： https://stackabuse.com/python-metaclasses-and-metaprogramming/

Answer 2

是的，正如您在自己的答案中输入的那样，使用元类来强制在特定方法上使用装饰器是正确的。通过一些更改，可以使要装饰的方法不固定 - 例如，在装饰函数中设置的属性可以用作“标记”，这样的装饰器应该强制覆盖方法。

除此之外，从 Python 3.6 开始，还有一种新的类级别机制 - 特殊方法 __init_subclass__，其特定目标是减少对元类的需求。元类可能很复杂，如果您的类层次结构需要组合多个元类，您可能会有些头疼。

__init_subclass__ 方法放在基类上，每次创建子类时调用一次。包装逻辑可以放在那里。

基本上，您可以修改您的装饰器以放置我上面提到的标记，并将这个类添加到您的继承层次结构中 - 它可以作为混合类放在多重继承中，因此可以重用于各种类树，如果需要：

def simple_decorator(func):
    def wrapper(*args, **kwargs):
        print("check cache")
        rt = func(*args, **kwargs)
        print("set cache")
        return rt
    wrapper.inherit_decorator = simple_decorator
    return wrapper

class InheritDecoratorsMixin:
    def __init_subclass__(cls, *args, **kwargs):
         super().__init_subclass__(*args, **kwargs)
         decorator_registry = getattr(cls, "_decorator_registry", {}).copy()
         cls._decorator_registry = decorator_registry
         # Check for decorated objects in the mixin itself- optional:
         for name, obj in __class__.__dict__.items():
              if getattr(obj, "inherit_decorator", False) and not name in decorator_registry:
                  decorator_registry[name] = obj.inherit_decorator
         # annotate newly decorated methods in the current subclass:
         for name, obj in cls.__dict__.items():
              if getattr(obj, "inherit_decorator", False) and not name in decorator_registry:
                  decorator_registry[name] = obj.inherit_decorator
         # finally, decorate all methods anottated in the registry:
         for name, decorator in decorator_registry.items():
              if name in cls.__dict__ and getattr(getattr(cls, name), "inherit_decorator", None) != decorator:
                    setattr(cls, name, decorator(cls.__dict__[name]))

所以，就是这样 - 每个新的子类都有自己的 _decorator_registry 属性，其中所有祖先中装饰方法的名称，以及要应用的 which 装饰器被注释。

如果装饰器应该为该方法使用一次，并且当被覆盖的方法对其祖先执行super() 调用时不重复（当您为缓存进行装饰时不是这种情况，因为超级方法赢了'not be called) 变得更棘手 - 但可以做到。

但是，这样做很棘手 - 因为超类中的装饰器实例将是子类上的装饰器之外的其他实例 - 一种将信息传递给“此方法的装饰器代码已在此链中运行”的方法call" 是使用实例级标记 - 如果代码要支持并行性，它应该是线程局部变量。

所有这些检查都会导致一些相当复杂的样板文件被放入一个简单的装饰器中——因此我们可以为我们想要运行一次的“装饰器”创建一个“装饰器”。换句话说，装饰有childmost bellow 的装饰器只会在“childmost”类上运行，而不是在调用super()时在超类中的相应方法上运行

import threading

def childmost(decorator_func):

    def inheritable_decorator_that_runs_once(func):
        decorated_func = decorator_func(func)
        name = func.__name__
        def wrapper(self, *args, **kw):
            if not hasattr(self, f"_running_{name}"):
                setattr(self, f"_running_{name}", threading.local())
            running_registry = getattr(self, f"_running_{name}")
            try:
                if not getattr(running_registry, "running", False):
                    running_registry.running = True
                    rt = decorated_func(self, *args, **kw)
                else:
                    rt = func(self, *args, **kw)
            finally:
                running_registry.running = False
            return rt

        wrapper.inherit_decorator = inheritable_decorator_that_runs_once
        return wrapper
    return inheritable_decorator_that_runs_once

---

使用第一个清单的示例：

class A: pass

class B(A, InheritDecoratorsMixin):
    @simple_decorator
    def method(self):
        print(__class__, "method called")

class C(B):
   def method(self):
       print(__class__, "method called")
       super().method()

在粘贴listing-1和这些A=B-C类之后 解释器，结果是这样的：

In [9]: C().method()                                                                         
check cache
<class '__main__.C'> method called
check cache
<class '__main__.B'> method called
set cache
set cache

（这里的“A”类完全是可选的，可以省略）

---

使用第二个清单的示例：

# Decorating the same decorator above:

@childmost
def simple_decorator2(func):
    def wrapper(*args, **kwargs):
        print("check cache")
        rt = func(*args, **kwargs)
        print("set cache")
        return rt
    return wrapper

class D: pass

class E(D, InheritDecoratorsMixin):
    @simple_decorator2
    def method(self):
        print(__class__, "method called")

class F(E):
   def method(self):
       print(__class__, "method called")
       super().method()

结果：

In [19]: F().method()                                                                        
check cache
<class '__main__.F'> method called
<class '__main__.E'> method called
set cache