如何输入动态创建的类，以便 mypy 可以正确地检查它们

Question

我正在寻求重构我贡献的名为 Pydra 的数据流引擎的函数任务装饰器，以便可以使用 mypy 对参数类型进行检查。基本上，我想在工作流构建时捕获函数参数并将它们存储在动态创建的

Inputs

类（使用python-attrs设计）中，这样我们就可以将函数的执行延迟到运行时。

这是我目前得到的代码

import typing as ty
import attrs
import inspect
from functools import wraps


@attrs.define(kw_only=True, slots=False)
class FunctionTask:

    name: str
    inputs: ty.Type[ty.Any]  # Use typing.Type hint for inputs attribute

    def __init__(self, name: str, **kwargs):
        self.name = name
        self.inputs = type(self).Inputs(**kwargs)


def task(function: ty.Callable) -> ty.Type[ty.Any]:

    sig = inspect.signature(function)

    inputs_dct = {
        p.name: attrs.field(default=p.default) for p in sig.parameters.values()
    }
    inputs_dct["__annotations__"] = {
        p.name: p.annotation for p in sig.parameters.values()
    }

    @wraps(function, updated=())
    @attrs.define(kw_only=True, slots=True, init=False)
    class Task(FunctionTask):
        func = staticmethod(function)

        Inputs = attrs.define(type("Inputs", (), inputs_dct))  # type: ty.Any

        inputs: Inputs = attrs.field()

        def __call__(self):
            return self.func(
                **{
                    f.name: getattr(self.inputs, f.name)
                    for f in attrs.fields(self.Inputs)
                }
            )

    return Task

你可以使用

@task
def myfunc(x: int, y: int) -> int:
    return x + y

# Would be great if `x` and `y` arguments could be type-checked as ints
mytask = myfunc(name="mytask", x=1, y=2)

# Would like mypy to know that mytask has an inputs attribute and that it has an int
# attribute 'x', so the linter picks up the incorrect assignment below
mytask.inputs.x = "bad-value"

mytask()

我想让 mypy 知道 mytask 有一个 inputs 属性并且它有一个 int 属性 'x'，因此 linter 会选择错误分配的“坏值”。如果可能的话，如果

myfunc.__init__

的关键字参数也进行了类型检查，那就太好了。

这可能吗？关于尝试的任何提示？

编辑：为了让我想做的更清楚一点，这里有一个动态生成的类如果静态编写的话会是什么样子的例子

@attrs.define
class StaticTask:

    @attrs.define
    class Inputs:
        x: int
        y: int

    name: str
    inputs: Inputs = attrs.field()

    @staticmethod
    def func(x: int, y: int) -> int:
        return x + y

    def __init__(self, name: str, x: int, y: int):
        self.name = name
        self.inputs.x = x
        self.inputs.y = y

    def __call__(self):
        return self.func(x=self.inputs.x, y=self.inputs.y)

Answer 1

我可能误解了你的需要，但如果你想做的只是向函数添加一个额外的类型化参数，你可以使用

typing.ParamSpec

.

P = ty.ParamSpec('P')
RV = ty.TypeVar('RV')


def task(f: ty.Callable[P, RV]) -> ty.Callable[ty.Concatenate[str, P], RV]:
    def _(name: str, **kwargs: P.kwargs) -> RV:
        # do something with name?
        return f(**kwargs)
    return _