具有 yield 的 Python 函数的正确类型注释
Proper type annotation of Python functions with yield
阅读 Eli Bendersky 的文章后 on implementing state machines via Python coroutines 我想...
- 在 Python3
下查看他的示例 运行
- 并为生成器添加适当的类型注释
我成功完成了第一部分(但没有使用 async def
s 或 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
...还有类型检查:
$ mypy --disallow-untyped-defs decoder.py
$
但我很确定我可以做得比在类型规范中仅使用 Generator
基础 class 更好(就像我对 Callable
所做的那样)。我知道它需要 3 个类型参数 (Generator[A,B,C]
),但我不确定它们在此处的具体指定方式。
欢迎任何帮助。
我自己想出了答案。
我进行了搜索,但在 official typing documentation for Python 3.5.2 中找不到 Generator
的 3 种类型参数的文档 - 除了真正神秘的提及...
class typing.Generator(Iterator[T_co], Generic[T_co, T_contra, V_co])
幸运的是,the original PEP484(开始了这一切)更有帮助:
》return类型的生成器函数可以通过Generator[yield_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
抱怨并要求正确的类型(如上所示)。
我会把这个问题搁置几天,以防有人想插话——尤其是在使用 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 documentation 也明确提到了如何处理异步情况(已接受的答案中已经提到了非异步示例)。
从那里引用:
async def echo_round() -> AsyncGenerator[int, float]:
sent = yield 0
while sent >= 0.0:
rounded = await round(sent)
sent = yield rounded
(第一个参数是 yield-type,第二个是 send-type)或者对于简单的情况(其中 send-type 是 None)
async def infinite_stream(start: int) -> AsyncIterator[int]:
while True:
yield start
start = await increment(start)
阅读 Eli Bendersky 的文章后 on implementing state machines via Python coroutines 我想...
- 在 Python3 下查看他的示例 运行
- 并为生成器添加适当的类型注释
我成功完成了第一部分(但没有使用 async def
s 或 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
...还有类型检查:
$ mypy --disallow-untyped-defs decoder.py
$
但我很确定我可以做得比在类型规范中仅使用 Generator
基础 class 更好(就像我对 Callable
所做的那样)。我知道它需要 3 个类型参数 (Generator[A,B,C]
),但我不确定它们在此处的具体指定方式。
欢迎任何帮助。
我自己想出了答案。
我进行了搜索,但在 official typing documentation for Python 3.5.2 中找不到 Generator
的 3 种类型参数的文档 - 除了真正神秘的提及...
class typing.Generator(Iterator[T_co], Generic[T_co, T_contra, V_co])
幸运的是,the original PEP484(开始了这一切)更有帮助:
》return类型的生成器函数可以通过Generator[yield_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
抱怨并要求正确的类型(如上所示)。
我会把这个问题搁置几天,以防有人想插话——尤其是在使用 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 documentation 也明确提到了如何处理异步情况(已接受的答案中已经提到了非异步示例)。
从那里引用:
async def echo_round() -> AsyncGenerator[int, float]:
sent = yield 0
while sent >= 0.0:
rounded = await round(sent)
sent = yield rounded
(第一个参数是 yield-type,第二个是 send-type)或者对于简单的情况(其中 send-type 是 None)
async def infinite_stream(start: int) -> AsyncIterator[int]:
while True:
yield start
start = await increment(start)