Proper type annotation of Python functions with yield

Question

After reading Eli Bendersky's article on implementing state machines via Python coroutines I wanted to...

• see his example run under Python3
• and also add the appropriate type annotations for the generators

I succeeded in doing the first part (but without using async defs or yield froms, I basically just ported the code - so any improvements there are most welcome).

But I need some help with the type annotations of the coroutines:

#!/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)

This runs properly...

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

...and also typechecks:

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

But I am pretty sure I can do better than just use the Generator base class in the type specs (just as I did for the Callable). I know it takes 3 type parameters (Generator[A,B,C]), but I am not sure how exactly they'd be specified here.

Any help most welcome.

Answer 1

I figured out the answer on my own.

I searched, but found no documentation for the 3 type parameters of Generator in the official typing documentation for Python 3.5.2 - beyond a truly cryptic mention of...

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

Luckily, the original PEP484 (that started all this) was far more helpful:

"The return type of generator functions can be annotated by the generic type Generator[yield_type, send_type, return_type] provided by typing.py module:

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

Based on this, I was able to annotate my Generators, and saw mypy confirm my assignments:

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:
    ...

I tested my assignments by e.g. swaping the order of the types - and then as expected, mypy complained and asked for the proper ones (as seen above).

The complete code is accessible from here.

I will leave the question open for a couple of days, in case anyone wants to chime in - especially in terms of using the new coroutine styles of Python 3.5 (async def, etc) - I would appreciate a hint on exactly how they'd be used here.

Answer 2

If you have a simple function using yield then you can use the Iterator type to annotate its result, rather than using Generator:

from collections.abc import Iterator  # Python >=3.9

def count_up() -> Iterator[int]:
    for x in range(10):
        yield x

If the function is async you need to use AsyncIterator instead:

from collections.abc import AsyncIterator  # Python >=3.5

async def count_up() -> AsyncIterator[int]:
    for x in range(10):
        yield x

In Python <3.9 you must import Iterator differently:

from typing import Iterator  # Python <3.9

Answer 3

As time of writing, the Python documentation explicitly mentions what to do with the async case as well (the non-async examples were already mentioned in the accepted answer).

Citing from there:

async def echo_round() -> AsyncGenerator[int, float]:
    sent = yield 0
    while sent >= 0.0:
        rounded = await round(sent)
        sent = yield rounded

(first parameter is yield-type, second is send-type) or for simple cases (where send-type is None)

async def infinite_stream(start: int) -> AsyncIterator[int]:
    while True:
        yield start
        start = await increment(start)