Рассмотрим эти две функции:
@asyncio.coroutine
async def get_int() -> int:
await asyncio.sleep(0.5)
return 5
@asyncio.coroutine
async def get_str() -> str:
await asyncio.sleep(0.5)
return "str"
Я хотел бы получить первую готовую сопрограмму и ее результат, поэтому я выполняю FIRST_COMPLETED
await:
next_int_coro = asyncio.Task(get_int())
next_str_coro = asyncio.Task(get_str())
done, pending = await asyncio.wait([next_int_coro, next_str_coro],
return_when=asyncio.FIRST_COMPLETED)
if next_int_coro in done:
some_int = await next_int_coro # type hint for some_int = Any :(
print(f"int : {some_int}")
elif next_str_coro in done:
some_str = await next_str_coro # type hint for some_str = Any :(
print(f"str : {some_str}")
else:
print("unexpected coroutine ended")
Проблема в том, что я хотел бы иметь подсказки типа при обработке результата задач. Вместо этого я получаю Any
.
Вместо того, чтобы оборачивать сопрограмму в Task
, я хорошо получаю подсказку типа, если ее ждут напрямую:
next_int_coro = get_int()
some_int = await next_int_coro # some_int is hinted to be an int
Но в этом случае , Я не могу определить, какая сопрограмма завершилась первой (if elif else
, который проверяет, принадлежит ли сопрограмма к набору «done
» Future
, попадает в else
).
Как я могу получить лучшее из обоих миров? Я имею в виду получение подсказок типа ожидаемого результата и с возможностью узнать, какая сопрограмма завершилась первой.