如何定义 ContextManager 协议

Question

我尝试使用类型提示来指定实现连接器类（在本例中为代理）时要遵循的 API。

我想指定这样的类应该是上下文管理器

我怎么做？

让我更清楚地重新表述一下：我如何定义Broker类，以便它表明它的具体实现（例如Rabbit类）必须是上下文管理器？

有没有实用的办法呢？我是否必须指定__enter__and__exit__并继承自Protocol？

继承就足够了吗ContextManager？

顺便问一下，我应该使用@runtimeor@runtime_checkable吗？（我的 VScode linter 似乎在查找 . 中的内容时遇到问题typing。我使用的是 python 3 7.5）

我知道如何使用 ABC 来做到这一点，但我想学习如何使用协议定义来做到这一点（我已经很好地使用了协议定义，但它们不是上下文管理器）。

我不知道如何使用该ContextManager类型。到目前为止我还没能从官方文档中找到好的例子。

目前我想出了

from typing import Protocol, ContextManager, runtime, Dict, List


@runtime
class Broker(ContextManager):
    """
    Basic interface to a broker.
    It must be a context manager
    """

    def publish(self, data: str) -> None:
        """
        Publish data to the topic/queue
        """
        ...

    def subscribe(self) -> None:
        """
        Subscribe to the topic/queue passed to constructor
        """
        ...

    def read(self) -> str:
        """
        Read data from the topic/queue
        """
        ...

和实施是

@implements(Broker)
class Rabbit:
    def __init__(self,
            url: str,
            queue: str = 'default'):
        """
        url: where to connect, i.e. where the broker is
        queue: the topic queue, one only
        """
        # self.url = url
        self.queue = queue
        self.params = pika.URLParameters(url)
        self.params.socket_timeout = 5

    def __enter__(self):
        self.connection = pika.BlockingConnection(self.params) # Connect to CloudAMQP
        self.channel = self.connection.channel() # start a channel
        self.channel.queue_declare(queue=self.queue) # Declare a queue
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.connection.close()

    def publish(self, data: str):
        pass  # TBD

    def subscribe(self):
        pass  # TBD

    def read(self):
        pass  # TBD

注意：implements装饰器工作正常（它来自以前的项目），它检查该类是给定协议的子类

Answer 1

简短的回答——你的Rabbit实现实际上是很好的。只需添加一些类型提示来指示 that__enter__返回其自身的实例并__exit__返回None. 参数的类型__exit__实际上并不重要。

更长的答案：

每当我不确定某个类型到底是什么/某个协议是什么时，检查 TypeShed 通常会很有帮助，它是标准库（以及一些第三方库）的类型提示集合。

例如，以下是 Typing.ContextManager 的定义。我已将其复制到下面：

from types import TracebackType

# ...snip...

_T_co = TypeVar('_T_co', covariant=True)  # Any type covariant containers.

# ...snip...

@runtime_checkable
class ContextManager(Protocol[_T_co]):
    def __enter__(self) -> _T_co: ...
    def __exit__(self, __exc_type: Optional[Type[BaseException]],
                 __exc_value: Optional[BaseException],
                 __traceback: Optional[TracebackType]) -> Optional[bool]: ...

读完这篇文章，我们知道了一些事情：

1. 该类型是一个协议，这意味着任何恰好实现__enter__并__exit__遵循上面给定签名的类型都将是有效的子类型，typing.ContextManager而无需显式继承它。

2. 这种类型是运行时可检查的，这意味着isinstance(my_manager, ContextManager)如果您出于某种原因想要这样做，那么做也可以。

3. 的参数名称__exit__均以两个下划线为前缀。这是类型检查器用于指示这些参数仅是位置的约定：使用关键字参数 on__exit__不会进行类型检查。实际上，这意味着您可以__exit__随意命名自己的参数，同时仍然遵守协议。

因此，将它们放在一起，这是仍然进行类型检查的 ContextManager 的最小可能实现：

from typing import ContextManager, Type, Generic, TypeVar

class MyManager:
    def __enter__(self) -> str:
        return "hello"

    def __exit__(self, *args: object) -> None:
        return None

def foo(manager: ContextManager[str]) -> None:
    with manager as x:
        print(x)        # Prints "hello"
        reveal_type(x)  # Revealed type is 'str'

# Type checks!
foo(MyManager())



def bar(manager: ContextManager[int]) -> None: ...

# Does not type check, since MyManager's `__enter__` doesn't return an int
bar(MyManager())

一个不错的小技巧是，__exit__如果我们实际上不打算使用参数，那么我们实际上可以使用相当懒惰的签名。毕竟，如果__exit__基本上接受任何内容，就不存在类型安全问题。

（更正式地说，符合 PEP 484 的类型检查器将尊重函数相对于其参数类型是逆变的）。

当然，如果您愿意，您可以指定完整的类型。例如，以您的Rabbit实现为例：

# So I don't have to use string forward references
from __future__ import annotations
from typing import Optional, Type
from types import TracebackType

# ...snip...

@implements(Broker)
class Rabbit:
    def __init__(self,
            url: str,
            queue: str = 'default'):
        """
        url: where to connect, i.e. where the broker is
        queue: the topic queue, one only
        """
        # self.url = url
        self.queue = queue
        self.params = pika.URLParameters(url)
        self.params.socket_timeout = 5

    def __enter__(self) -> Rabbit:
        self.connection = pika.BlockingConnection(params) # Connect to CloudAMQP
        self.channel = self.connection.channel() # start a channel
        self.channel.queue_declare(queue=self.queue) # Declare a queue
        return self

    def __exit__(self,
                 exc_type: Optional[Type[BaseException]],
                 exc_value: Optional[BaseException],
                 traceback: Optional[TracebackType],
                 ) -> Optional[bool]:
        self.connection.close()

    def publish(self, data: str):
        pass  # TBD

    def subscribe(self):
        pass  # TBD

    def read(self):
        pass  # TBD

回答新编辑的问题：

如何定义 Broker 类，以便它指示其具体实现（例如 Rabbit 类）必须是上下文管理器？

有没有实用的办法呢？我是否必须指定进入和退出并仅从协议继承？

继承ContextManager就够了吗？

有两种方法：

1. 重新定义__enter__和__exit__函数，从 ContextManager 复制原始定义。
2. 让 Broker 成为ContextManager 和 Protocol 的子类。

如果您仅子类化 ContextManager，那么您所做的就是让 Broker 或多或少地继承 ContextManager 中恰好有默认实现的任何方法。

PEP 544：协议和结构类型对此进行了更多详细介绍。协议上的 mypy 文档有一个更用户友好的版本。例如，请参阅有关定义子协议和子类化协议的部分。

顺便问一下，我应该使用@runtime还是@runtime_checkable？（我的 VScode linter 似乎无法在输入中找到这些内容。我使用的是 python 3 7.5）

它应该是runtime_checkable。

也就是说，Protocol 和 runtime_checkable 实际上都是在 3.8 版本中添加到 Python 中的，这可能就是你的 linter 不满意的原因。

如果您想在旧版本的 Python 中使用两者，则需要 pip install Typing-extensions，这是类型类型的官方向后移植。

一旦安装完毕，您就可以执行from typing_extensions import Protocol, runtime_checkable.