Uh oh!
There was an error while loading.Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork33.7k
Closed
Description
Bug report
Bug description:
asyncio/staggered.py is usingtyping.Optional:
cpython/Lib/asyncio/staggered.py
Line 73 in65de194
| previous_failed:typing.Optional[locks.Event])->None: |
However,#114281 removed thetyping import. This causes test failures inaiohappyeyeballs:
coro_fns = <generator object start_connection.<locals>.<genexpr> at 0x7feba06d1be0>, delay = 0.3asyncdefstaggered_race(coro_fns,delay,*,loop=None):"""Run coroutines with staggered start times and take the first to finish. This method takes an iterable of coroutine functions. The first oneis started immediately. From then on, whenever the immediately preceding one fails (raises an exception),or when*delay* seconds has passed, thenext coroutineis started. This continues until one of the coroutines complete successfully,in which caseall others are cancelled,or untilall coroutines fail. The coroutines provided should be well-behavedin the following way:* They should only``return``if completed successfully.* They should alwaysraise an exceptionif they didnot complete successfully. In particular,if they handle cancellation, they should probably reraise, like this::try:# do workexcept asyncio.CancelledError:# undo partially completed workraise Args: coro_fns: an iterable of coroutine functions, i.e. callables thatreturn a coroutineobject when called. Use``functools.partial``or lambdas topass arguments. delay: amount of time,in seconds, between starting coroutines. If``None``, the coroutines will run sequentially. loop: the event loop to use. Returns:tuple*(winner_result, winner_index, exceptions)* where-*winner_result*: the result of the winning coroutine,or``None``if no coroutines won.-*winner_index*: the index of the winning coroutinein``coro_fns``,or``None``if no coroutines won. If the winning coroutine mayreturnNone on success,*winner_index* can be used to definitively determine whetherany coroutine won.-*exceptions*:list of exceptions returned by the coroutines.``len(exceptions)``is equal to the number of coroutines actually started,and the orderis the sameasin``coro_fns``. The winning coroutine's entry is ``None``."""#TODO: when we have aiter() and anext(), allow async iterables in coro_fns. loop= loopor events.get_running_loop() enum_coro_fns=enumerate(coro_fns) winner_result=None winner_index=None exceptions= [] running_tasks= []asyncdefrun_one_coro(> previous_failed: typing.Optional[locks.Event]) -> None:E NameError: name 'typing' is not defined. Did you forget to import 'typing'?coro_fns = <generator object start_connection.<locals>.<genexpr> at 0x7feba06d1be0>delay = 0.3enum_coro_fns = <enumerate object at 0x7feba06223e0>exceptions = []loop = <_UnixSelectorEventLoop running=False closed=False debug=False>running_tasks = []winner_index = Nonewinner_result = None/usr/lib/python3.13/asyncio/staggered.py:73: NameError
CPython versions tested on:
3.13, CPython main branch
Operating systems tested on:
Linux
Linked PRs
Metadata
Metadata
Assignees
Projects
Status
Done