Python 函数的正确类型注释及其产生式

在读了 Eli Bendersky 的文章 通过 Python 协程实现状态机之后,我想..。

  • 在 Python 3下运行他的示例
  • 并为生成器添加适当的类型注释

我成功地完成了第一部分(但是没有使用 ABC0或 yield from,我基本上只是移植了代码-所以任何改进都是非常受欢迎的)。

但是在协程的类型注释方面我需要一些帮助:

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

还有打字检查:

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

但是我非常肯定我可以做得比仅仅在类型规格中使用 Generator基类更好(就像我对 Callable所做的那样)。我知道它需要3个类型参数(Generator[A,B,C]) ,但我不确定它们在这里具体是如何指定的。

欢迎任何帮助。

29144 次浏览

我自己找到了答案。

我搜索,但没有发现文档的3类型参数的 GeneratorPython 3.5.2的官方类型化文档-以外的真正神秘的提及..。

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

幸运的是,最初的 PEP484(这一切的始作俑者)的帮助要大得多:

”生成器函数的返回类型可以由 typeing.py 模块提供的泛型类型 Generator [ production _ type,send _ type,return _ type ]进行注释:

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抱怨并要求正确的类型(如上所示)。

完整代码 可以从这里进入

我会把这个问题保留几天,以防有人想插嘴——特别是在使用 Python 3.5(async def等)的新协程风格方面——我希望能得到一个关于它们在这里具体如何使用的提示。

如果您有一个使用 yield的简单函数,那么您可以使用 Iterator类型来注释它的结果,而不是使用 Generator:

from typing import Iterator


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

在撰写本文时,Python 文档也明确提到了如何处理异步情况(在已接受的答案中已经提到了非异步示例)。

引用如下:

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

(第一个参数是屈服类型,第二个参数是发送类型)或简单情况下(其中发送类型是无)

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