惰性评估：将操作转发到递延值

Question

我已经实现了用于延迟评估转储到 JSON 的配置的类。没问题，只需扩展编码器以使用特定协议（固定方法/属性）主动评估类。

class DeferredCall(object):
  """Call that is evaluated lazyly"""
  def __init__(self, func, *func_args, **func_kwargs):
    self.func = func
    self.func_args = func_args
    self.func_kwargs = func_kwargs

  def resolve(self):  # called by JSON encoder
    return self.func(*self.func_args, **self.func_kwargs)

 a = DeferredCall(lambda: 1)
 a # gives <[module].DeferredCall at 0x1e99410>
 a.resolve() # gives 1

现在，随着强大的力量而来的是想要更多力量的用户。也就是说，直接对类而不是它们代表的值进行操作。根据python数据模型，这应该像实现魔法方法一样简单，例如

__add__

，

__len__

等

添加

def __add__(self, other):
    return self.resolve() + other

或

def __add__(self, other):
    return self.resolve().__add__(other)

会正确地给我

a + 3 == 4

。

不过，实现

all魔法方法有点太多了。所以我尝试使用__getattr__

def __getattr__(self, item):
  return getattr(self.resolve(), item)

适用于

a.__mul__(3) == 3

，但使用 

a * 3 == 3

 会爆炸 

TypeError: unsupported operand type(s) for *: 'DeferredCall' and 'int'

。

那么有没有什么

其他方法可以将运算符转发到包装值？理想情况下，无需冒险以编程方式编写代码，也无需经历 __getattribute__

 的麻烦。

Answer 1

发布我的解决方案，以防其他人需要。根据设计，大多数“内置”操作（例如

*

 或 

len

）不会使用 

__getattr[ibute]__

。

我已经决定以编程方式创建方法

class DeferredCall(object):
  def __init__(self, func, *func_args, **func_kwargs):
    self.func = func
    self.func_args = func_args
    self.func_kwargs = func_kwargs

  def resolve(self):
    return self.func(*self.func_args, **self.func_kwargs)

  # magic must be resolved explicitly
  # self other - resolve in reflected order to allow other to resolve as well
  for so_magic in [("lt", "gt"), ("le", "ge"), ("eq", "eq"), ("ne", "ne"), ("add", "radd"), ("sub", "rsub"), ("mul", "rmul"), ("div", "rdiv"), ("truediv", "rtruediv"), ("floordiv", "rfloordiv"), ("mod", "rmod"), ("divmod", "rdivmod"), ("pow", "rpow"), ("lshift", "rlshift"), ("rshift", "rrshift"), ("and", "rand"), ("xor", "rxor"), ("or", "ror")]:
    for func_name, refl_name in [(so_magic[0], so_magic[1]), (so_magic[1], so_magic[0])]:
      exec("def __%(fname)s__(self, other):\n\ttry:\n\t\tres = other.__%(rname)s__(self.resolve())\n\t\tif res == NotImplemented:\n\t\t\traise AttributeError\n\texcept AttributeError:\n\t\tres = self.resolve().__%(fname)s__(other)\n\treturn res" % {"fname": func_name, "rname": refl_name})

  # regular magic - immutable only
  for magic in ("str", "nonzero", "unicode", "getattr", "call", "len", "getitem", "missing", "iter", "reversed", "contains", "getslice", "neg", "pos", "abs", "invert", "complex", "int", "long", "float", "oct", "hex", "index"):
    exec("def __%(fname)s__(self, *args, **kwargs):\n\treturn self.resolve().__%(fname)s__(*args, **kwargs)" % {"fname": magic})

基本上，魔术方法必须分为两类：独立方法和上下文方法。

独立的创建起来很简单，解决调用并执行魔术方法。例如，

len

解析为：

def __len__(self, *args, **kwargs):
  return self.resolve().__len__(*args, **kwargs)

上下文必须颠倒调用，例如“大于”

other

 实际上检查 

other

 是否“小于”

self

。如果两个对象都是延迟调用，则这是必需的，从而允许 

other

 自行解析；否则，很多方法都会引发

TypeError

。仅当 

other

 没有反转版本时才使用直接评估。

def __gt__(self, other):
  try:
    res = other.__lt__(self.resolve())
    if res == NotImplemented:
      raise AttributeError
  except AttributeError:
    res = self.resolve().__gt__(other)
  return res

有些调用可能可以更有效地实现，因为 python 使用了一些技巧（这正是我的问题首先出现的地方）。例如，乘法可以利用交换性：

def __mul__(self, other):
  """self * other"""
  return other * self.resolve()

Answer 2

您可以使用元类来动态生成所有函数。您可以显式指定您希望实现的所有方法的名称，也可以检查现有类并在您的类上创建具有相同名称的方法。

例如，创建

int

实现的所有方法：

def _wrapper(func_name):
    def method(self, *args, **kwargs):
        value = self.resolve()
        func = getattr(value, func_name, None)
        if func is None:
            raise AttributeError(func_name)
        resolved_args = [
            arg.resolve() if isinstance(arg, DeferredCall) else arg
            for arg in args
        ]
        resolved_kwargs = {
            arg_name: arg.resolve() if isinstance(arg, DeferredCall) else arg
            for arg_name, arg in kwargs.items()
        }
        return func(*resolved_args, **resolved_kwargs)
    return method


class DeferredCallMetaclass(type):
    NON_WRAPPED_METHODS = (
        "__class__",
        "__class_getitem__",
        "__delattr__",
        "__dir__",
        "__doc__",
        "__getattribute__",
        "__getstate__",
        "__init__",
        "__init_subclass__",
        "__new__",
        "__repr__",
        "__setattr__",
        "__subclasshook__",
    )

    def __new__(cls, clsname, bases, attrs, **additional_kwargs):
        # Find all the routines belonging to the int class and create corresponding
        # wrapped methods on DeferredCall (except for the methods listed above).
        for attr, _ in getmembers(int, predicate=isroutine):
            if (
                attr not in attrs
                and attr not in DeferredCallMetaclass.NON_WRAPPED_METHODS
            ):
                attrs[attr] = _wrapper(attr)
        return super().__new__(cls, clsname, bases, attrs, **additional_kwargs)


class DeferredCall(object, metaclass=DeferredCallMetaclass):
    """Call that is evaluated lazyly"""
    def __init__(self, func, *func_args, **func_kwargs):
        self.func = func
        self.func_args = func_args
        self.func_kwargs = func_kwargs

    def resolve(self):  # called by JSON encoder
        return self.func(*self.func_args, **self.func_kwargs)

然后：

instance1 = DeferredCall( lambda x: x, 3)
instance2 = DeferredCall( lambda x, y, z: z, 2, 3, z=4)
print("instance1 + 1          ->", instance1 + 1)
print("2 ** instance1         ->", 2 ** instance1)
print("instance1 > instance2  ->", instance1 > instance2)
print("instance1 >= instance2 ->", instance1 >= instance2)
print("instance1 <= instance2 ->", instance1 <= instance2)
print("instance1 != instance2 ->", instance1 != instance2)

输出：

instance1 + 1          -> 4
2 ** instance1         -> 8
instance1 > instance2  -> False
instance1 >= instance2 -> False
instance1 <= instance2 -> True
instance1 != instance2 -> True