输入函数对象属性

问题描述 投票：0回答：1

我必须实现一个发送消息并依次获得响应的协议。通常，响应是即时的，但在某些情况下，消息将设备设置为正确的状态以处理其他消息，并且只有在处理其他消息后，设备才会响应设置状态的初始消息。

由于基于参数生成消息和构造响应对象的实际代码相当重复，因此我使用装饰器来执行此操作。虽然在 90% 的情况下用户想要 

sendMessageX() -> ResponseToMessageX
，但在某些情况下用户必须 
dispatchMessageX() -> Future<ResponseToMessageX>
。由于这两个函数采用相同的参数并且本质上返回相同的东西（
ResponsToMessageX
，至少在解析 future 之后），因此从装饰器生成这两个函数是有意义的。

对我来说最简单的方法是滥用 python 

everything is an object
并简单地将第二个函数作为属性添加到函数对象中，即：

def decorator(f):
    def decorated_f():
        return f()

    def decorated_f_split():
        return f'()
    
    decorated_f.second_function = decorated_f_split

    return decorated_f

（如果我可以帮助的话，我会尝试避免元类以使事情变得简单并避免限制）。

为了方便用户使用，我希望输入提示仍然有效（以便编辑器可以输入提示响应对象具有哪些属性）。

当我让它工作时，pyright 仍然抱怨，我无法摆脱警告。这是我想出的 MWE：

from functools import wraps
from typing import (
    Any,
    Awaitable,
    Callable,
    Concatenate,
    ParamSpec,
    Protocol,
    TypeVar,
    cast,
)

T = TypeVar("T")
C = TypeVar("C", covariant=True)
P = ParamSpec("P")  # , covariant=True)


class X(Protocol[P, C]):
    def split(self, *args: Any, **kwds: Any) -> Awaitable[C]:
        ...

    def __call__(self, *args: Any, **kwds: Any) -> C:
        ...


def wrap_returntype_awaitable(f: Callable[P, T]) -> Callable[P, Awaitable[T]]:
    f.__annotations__["return"] = Awaitable[T]

    g = cast(Callable[P, Awaitable[T]], f)
    return g


def decorator(f: Callable[Concatenate[Any, P], C]) -> X[P, C]:
    class Wrapper:
        # @wrap_returntype_awaitable
        @wraps(wrap_returntype_awaitable(f))
        async def split(self, *args: P.args, **kwds: P.kwargs) -> Awaitable[C]:
            await self.send_command()
            return self.get_response()

        @wraps(f)
        async def __call__(self, *args: P.args, **kwds: P.kwargs) -> C:
            response_receiver = await Wrapper.split(self, *args, **kwds)
            return await response_receiver

    Wrapper.split.__annotations__["return"] = Awaitable[C]

    return Wrapper()


class A:
    async def send_command(self):
        pass

    async def get_response(self):
        pass

    async def command(self, a: int) -> int:
        await self.send_command()
        return a

    @decorator
    async def x(self, a1: int) -> int:
        ...


async def demo():
    a = A()

    response = await a.x()
    split_command = await a.x.split()
    split_response = await split_command

我觉得被迫在装饰器中使用一个类，因为我无法告诉 pytright 我想分配给该函数的属性，否则（我不知道如何告诉 Pyright 我返回的对象有一个属性

split
我需要设置（除了抑制该行上的类型警告））。问题在于匹配 
self
类型，特别是因为我通常从 
send_command
的基本类型继承 
get_response
/
class A
消息（省略以保持 MWE 更简单）。

Pyright 抱怨以下内容：

p2.py:38:24 - error: Cannot access member "send_command" for type "Wrapper"        
    Member "send_command" is unknown (reportGeneralTypeIssues)
p2.py:39:25 - error: Cannot access member "get_response" for type "Wrapper"        
    Member "get_response" is unknown (reportGeneralTypeIssues)
p2.py:48:12 - error: Expression of type "Wrapper" cannot be assigned to return type "X[P@decorator, C@decorator]"
    "Wrapper" is incompatible with protocol "X[P@decorator, C@decorator]"
      Type parameter "P@X" is invariant, but "P@X" is not the same as "P@decorator"
      Type parameter "C@X" is covariant, but "Awaitable[C@decorator]" is not a subtype of "C@decorator"
        Type "Awaitable[C@decorator]" cannot be assigned to type "C@decorator" (reportGeneralTypeIssues)
3 errors, 0 warnings, 0 informations

我已经触及了第一个警告（我应该如何在这里输入 self？将其强制为 

Base_A
不起作用，（
Base_A
有 
send_command
/
get_response
，但我的装饰器用 
A._marker_property
来调用它们） .此外，由于我当前使用的是函数而不是类装饰器，因此我无法到达装饰器内部的
class A
。

我可以尝试将装饰器拆分为一个标记我的函数的函数装饰器和一个类装饰器（可以访问底层类），但随后我会再次抱怨函数属性的分配。

我更大的问题是最后两个错误：

    "Wrapper" is incompatible with protocol "X[P@decorator, C@decorator]"
      Type parameter "P@X" is invariant, but "P@X" is not the same as "P@decorator"
      Type parameter "C@X" is covariant, but "Awaitable[C@decorator]" is not a subtype of "C@decorator"
        Type "Awaitable[C@decorator]" cannot be assigned to type "C@decorator"

为什么

P@X
与
P@decorator
不一样（参数集应该相同？而且我不能有协变参数集...） 我认为我已经正确输入了它，因此 
Awaitable[C@decorator]
应该出现在正确的位置，但类型检查器显然有不同的想法。

（我无法控制协议，也无法更改其设计）

编辑：我非常专注于完成/推理工作（即，对于示例，编辑器知道函数返回

int
/
Awaitable[int]
），以至于我没有实际测试该功能。 self 的访问中断，父对象的方法无法调用:/

python type-hinting pyright
1个回答
0
投票

虽然解决方案并不理想，并且添加了更多约束，但这就是我想出的（它使类型检查器满意，并且类型推断有效）：

from functools import wraps
from typing import (
    Any,
    Awaitable,
    Callable,
    Concatenate,
    ParamSpec,
    Protocol,
    TypeVar,
    cast,
    runtime_checkable,
)

T = TypeVar("T")
C = TypeVar("C", covariant=True)
C2 = TypeVar("C2")
P = ParamSpec("P")  # , covariant=True)


class Response:
    ...


class Commandable(Protocol):
    async def send_command(self) -> Awaitable[Response]:
        ...


@runtime_checkable
class X(Protocol[P, C]):
    _commandable_parent: Commandable

    async def split(self, *args: Any, **kwds: Any) -> Awaitable[C]:
        ...

    def __call__(self, *args: Any, **kwds: Any) -> C:
        ...


def wrap_returntype_awaitable(f: Callable[P, T]) -> Callable[P, Awaitable[T]]:
    f.__annotations__["return"] = Awaitable[T]

    g = cast(Callable[P, Awaitable[T]], f)
    return g


def decorator(f: Callable[Concatenate[Any, P], C]) -> X[P, C]:
    class Wrapper(X[P, C2]):  # C2 should be C here, but pyright does not like it
        def __init__(self):
            self._commandable_parent = None  # type: ignore # the class decorator fixes this

        @wraps(wrap_returntype_awaitable(f))
        async def split(
            self, *args: P.args, **kwds: P.kwargs
        ) -> Awaitable[Response]:  # if I use Awaitable[C2] here pyright has issues
            return await self._commandable_parent.send_command()

        @wraps(f)
        async def __call__(
            self, *args: P.args, **kwds: P.kwargs
        ) -> Response:  # C2 and Response are not compatible in the return by pyright
            response_receiver = await Wrapper.split(self, *args, **kwds)
            return await response_receiver

    Wrapper.split.__annotations__["return"] = Awaitable[C]

    return Wrapper()


def fixup_parents(cls):
    initialize_command_parent: list[X] = []

    for attribute_name in dir(cls):
        attribute = getattr(cls, attribute_name)

        if isinstance(attribute, X):
            initialize_command_parent.append(attribute)

    original_init = getattr(cls, "__init__")

    @wraps(original_init)
    def fixed_init(self, *args, **kwargs):
        original_init(self, *args, **kwargs)

        for attribute in initialize_command_parent:
            attribute._commandable_parent = self

    cls.__init__ = fixed_init

    return cls


@fixup_parents
class A:
    async def send_command(self):
        return self.get_response()

    async def get_response(self):
        pass

    async def command(self, a: int) -> int:
        await self.send_command()
        return a

    @decorator
    async def x(self, a1: int) -> int:
        ...


async def demo():
    a = A()

    response = await a.x()
    split_command = await a.x.split()
    split_response = await split_command
    print("done")


import asyncio

asyncio.run(demo())

虽然这有效，但有几个难看的地方：

  1. 我未能在 
    Wrapper
    函数内为 
    decorator
    类重用 TypeVar C
  2. 必须修复自我（通过类装饰器
    fixup_parents
    替换方法的可调用类（我不知道更干净的解决方案））
  3. 包装器方法的返回类型是静态分配的，并且没有使用应有的泛型。 （我无法说服皮赖特）。虽然由于 
    @wraps
    和装饰器函数的返回类型而隐藏了它，但它肯定不干净

欢迎针对上述问题有更好的解决方案/方案。我什至可能将这些发布在一个新问题/pyright qa 网站上。

最新问题
© www.soinside.com 2019 - 2024. All rights reserved.