Proper type annotation of Python functions with yield
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.
If you have a simple function using yield
, then you can use the Iterator
type to annotate its result rather than Generator
:
from typing import Iteratordef count_up() -> Iterator[int]: for x in range(10): yield x
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)