Правильная аннотация типа функций Python с доходностью

После прочтения статьи Эли Бендерски по внедрению государственных машин с помощью сопрограмм Python я хотел...

  • см. его пример под управлением Python3
  • а также добавить соответствующие аннотации типов для генераторов

Мне удалось выполнить первую часть (но без использования async def или yield from s, я просто просто портировал код, поэтому любые улучшения там наиболее приветствуются).

Но мне нужна помощь с аннотациями типа сопрограммы:

#!/usr/bin/env python3

from typing import Callable, Generator

def unwrap_protocol(header: int=0x61,
                    footer: int=0x62,
                    dle: int=0xAB,
                    after_dle_func: Callable[[int], int]=lambda x: x,
                    target: Generator=None) -> Generator:
    """ Simplified protocol unwrapping co-routine."""
    #
    # Outer loop looking for a frame header
    #
    while True:
        byte = (yield)
        frame = []  # type: List[int]

        if byte == header:
            #
            # Capture the full frame
            #
            while True:
                byte = (yield)
                if byte == footer:
                    target.send(frame)
                    break
                elif byte == dle:
                    byte = (yield)
                    frame.append(after_dle_func(byte))
                else:
                    frame.append(byte)


def frame_receiver() -> Generator:
    """ A simple co-routine "sink" for receiving full frames."""
    while True:
        frame = (yield)
        print('Got frame:', ''.join('%02x' % x for x in frame))

bytestream = bytes(
    bytearray((0x70, 0x24,
               0x61, 0x99, 0xAF, 0xD1, 0x62,
               0x56, 0x62,
               0x61, 0xAB, 0xAB, 0x14, 0x62,
               0x7)))

frame_consumer = frame_receiver()
next(frame_consumer)  # Get to the yield

unwrapper = unwrap_protocol(target=frame_consumer)
next(unwrapper)  # Get to the yield

for byte in bytestream:
    unwrapper.send(byte)

Это работает правильно...

$ ./decoder.py 
Got frame: 99afd1
Got frame: ab14

... и также typechecks:

$ mypy --disallow-untyped-defs decoder.py 
$

Но я уверен, что могу сделать лучше, чем просто использовать базовый класс Generator в спецификациях типов (как и для Callable). Я знаю, что он принимает 3 параметра типа (Generator[A,B,C]), но я не уверен, как именно они будут указаны здесь.

Любая помощь приветствуется.

Ответы

Ответ 1

Я понял ответ самостоятельно.

Я искал, но не нашел документацию для трех типов параметров Generator в официальной документации по типу для Python 3.5.2 - за истинно загадочным упоминанием...

class typing.Generator(Iterator[T_co], Generic[T_co, T_contra, V_co])

К счастью, оригинальный PEP484 (который начал все это) был гораздо полезнее:

"Возвращаемый тип функций генератора может быть аннотирован генератором Generic [output_type, send_type, return_type], предоставленным модулем typing.py:

def echo_round() -> Generator[int, float, str]:
    res = yield
    while res:
        res = yield round(res)
    return 'OK'

Исходя из этого, я смог аннотировать мои генераторы и увидел mypy подтверждение моих назначений:

from typing import Callable, Generator

# A protocol decoder:
#
# - yields Nothing
# - expects ints to be `send` in his yield waits
# - and doesn't return anything.
ProtocolDecodingCoroutine = Generator[None, int, None]

# A frame consumer (passed as an argument to a protocol decoder):
#
# - yields Nothing
# - expects List[int] to be `send` in his waiting yields
# - and doesn't return anything.
FrameConsumerCoroutine = Generator[None, List[int], None]


def unwrap_protocol(header: int=0x61,
                    footer: int=0x62,
                    dle :int=0xAB,
                    after_dle_func: Callable[[int], int]=lambda x: x,
                    target: FrameConsumerCoroutine=None) -> ProtocolDecodingCoroutine:
    ...

def frame_receiver() -> FrameConsumerCoroutine:
    ...

Я проверил свои задания, например. поменяв порядок типов, а затем, как и ожидалось, mypy пожаловался и попросил правильные (как видно выше).

Полный код доступен здесь.

Я оставлю вопрос открытым на пару дней, если кто-то захочет перезвонить - особенно с точки зрения использования новых стилей coroutine для Python 3.5 (async def и т.д.) - я был бы признателен за подсказку как они будут использоваться здесь.