Как определить протокол ContextManager - PullRequest
0 голосов
/ 30 января 2020

Я пытаюсь использовать подсказку типа, чтобы указать API, которому следует следовать при реализации класса соединителя (в данном случае брокеру).

Я хочу указать, что такой класс (ы) должен быть контекстным менеджер (ы)

Как мне это сделать?

Позвольте мне перефразировать это более четко: как я могу определить класс Broker, чтобы он указывал, что его конкретные реализации, например, класс Rabbit, должны быть контекстными менеджерами?

Есть ли практический способ? Нужно ли указывать __enter__ и __exit__ и просто наследовать от Protocol?

Достаточно ли наследовать от ContextManager?

Кстати, мне следует использовать @runtime или @runtime_checkable? (Кажется, что у моего линкера VScode проблемы в typing. Я использую python 3 7.5)

Я знаю, как это сделать с AB C, но я хотел бы узнать как это сделать с определениями протоколов (которые я уже использовал нормально, но они не были менеджерами контекста).

Я не могу разобраться, как использовать тип 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 декоратор работает нормально (он приходит из предыдущего проекта), он проверяет, является ли класс подклассом данного протокола

1 Ответ

1 голос
/ 30 января 2020

Краткий ответ - ваша реализация 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__ не приведет к проверке типа. На практике это означает, что вы можете называть свои собственные параметры __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. Сделать брокер подклассом оба ContextManager и Protocol .

Если вы наследуете только ContextManager, все, что вы делаете, это заставляете брокера просто наследовать любые методы, имеющие реализацию по умолчанию в ContextManager, более или менее.

PEP 544: Протоколы и структурная типизация более подробно об этом. mypy docs по протоколам имеют более удобную версию этого. Например, см. Раздел Определение подпротоколов и протоколов подклассов .

Кстати, я должен использовать @runtime или @runtime_checkable? (Кажется, у моего линкера VScode проблемы с поиском при наборе текста. Я использую python 3 7.5)

Это должно быть runtime_checkable.

Тем не менее, как протокол, так и runtime_checkable были фактически добавлены в Python в версии 3.8, и, вероятно, именно поэтому ваш линтер недоволен.

Если вы захотите использовать оба в более старых версиях Python, вы необходимо установить pip typing-extensions , официальный бэкпорт для набора типов.

После того, как это установлено, вы можете сделать from typing_extensions import Protocol, runtime_checkable.

...