如何定义ContextManager协议

Question

我试图使用类型暗示来指定在实现连接器类时要遵循的API (在本例中是对代理)。

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

我该怎么做？

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

有实用的方法吗？我必须指定__enter__和__exit__并从Protocol继承吗？

从ContextManager继承就足够了吗

顺便问一下，我应该使用@runtime还是@runtime_checkable？(我的VScode linter似乎很难在typing中找到它们。我正在使用python 37.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实现实际上很好--就像现在一样。只需添加一些类型提示，以指示__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__的参数名都以两个下划线为前缀。这是一个常规类型检查器，用于指示那些参数仅为位置:在__exit__上使用关键字参数不会键入check。实际上，这意味着您可以在仍然遵守协议的情况下，任意命名您自己的__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())

一个很好的小技巧是，如果我们不打算使用params，那么我们实际上可以使用一个非常懒惰的__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类，以便它指示它的具体实现，例如兔子类，必须是上下文管理器？ 有实用的方法吗？我是否必须指定、输入和退出并从协议继承？ 它是否足以继承ContextManager？

有两种方式：

1. 重新定义__enter__和__exit__函数，从ContextManager复制原始定义。
2. 使代理子类同时包括ContextManager和协议。

如果您只使用ContextManager子类，那么您所做的就是让代理继承任何碰巧在ContextManager中具有默认实现的方法，或多或少。

PEP 544:协议和结构类型将详细介绍这方面的内容。关于协议的格式文档有一个更方便用户的版本。例如，请参阅关于定义子协议和子类协议的部分。

顺便问一下，我是使用@运行时还是使用@runtime_checkable？(我的VScode linter似乎在打字时找不到这些东西。我正在使用python 37.5)

应该是runtime_checkable。

也就是说，协议和runtime_checkable实际上都是在版本3.8中添加到Python中的，这可能就是linter不高兴的原因。

如果您想在较早版本的Python中同时使用这两种方法，您将需要安装打字-扩展，这是输入类型的官方支持端口。

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