如何定义ContextManager协议

Question

我正在尝试使用类型提示来指定在实现连接器类时（在这种情况下，对代理来说）要遵循的API。

我想指定此类应该是上下文管理器

我该怎么做？

让我更清楚地改写一下：我该如何定义Broker类，以便它指示其具体实现，例如Rabbit类，必须是上下文管理器吗？

是否有实用的方法？我是否必须指定__enter__和__exit__并仅继承自Protocol？

是否足以从ContextManager继承？

我知道如何使用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(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装饰器工作正常（它来自上一个项目），它检查该类是否为给定协议的子类

我正在尝试使用类型提示来指定实现连接器类（在这种情况下为代理）时要遵循的API。我想指定这样的类应该是上下文管理器。...

Answer 1

简短的答案-您的Rabbit实现实际上是按原样进行的。只需添加一些类型提示即可指示__enter__返回其自身的实例，并且__exit__返回None。 __exit__参数的类型实际上并不重要。