`asyncio.staggered` is missing `typing` import

[mgorny]

Bug report

Bug description:

asyncio/staggered.py is using typing.Optional:

cpython/Lib/asyncio/staggered.py

Line 73 in 65de194

previous_failed: typing.Optional[locks.Event]) -> None:

However, #114281 removed the typing import. This causes test failures in aiohappyeyeballs:

coro_fns = <generator object start_connection.<locals>.<genexpr> at 0x7feba06d1be0>, delay = 0.3

 async def staggered_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 one is
 started immediately. From then on, whenever the immediately preceding one
 fails (raises an exception), or when *delay* seconds has passed, the next
 coroutine is started. This continues until one of the coroutines complete
 successfully, in which case all others are cancelled, or until all
 coroutines fail.
 
 The coroutines provided should be well-behaved in the following way:
 
 * They should only ``return`` if completed successfully.
 
 * They should always raise an exception if they did not complete
 successfully. In particular, if they handle cancellation, they should
 probably reraise, like this::
 
 try:
 # do work
 except asyncio.CancelledError:
 # undo partially completed work
 raise
 
 Args:
 coro_fns: an iterable of coroutine functions, i.e. callables that
 return a coroutine object when called. Use ``functools.partial`` or
 lambdas to pass 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 coroutine in
 ``coro_fns``, or ``None`` if no coroutines won. If the winning
 coroutine may return None on success, *winner_index* can be used
 to definitively determine whether any coroutine won.
 
 - *exceptions*: list of exceptions returned by the coroutines.
 ``len(exceptions)`` is equal to the number of coroutines actually
 started, and the order is the same as in ``coro_fns``. The winning
 coroutine's entry is ``None``.
 
 """
 # TODO: when we have aiter() and anext(), allow async iterables in coro_fns.
 loop = loop or events.get_running_loop()
 enum_coro_fns = enumerate(coro_fns)
 winner_result = None
 winner_index = None
 exceptions = []
 running_tasks = []
 
 async def run_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.3
enum_coro_fns = <enumerate object at 0x7feba06223e0>
exceptions = []
loop = <_UnixSelectorEventLoop running=False closed=False debug=False>
running_tasks = []
winner_index = None
winner_result = None

/usr/lib/python3.13/asyncio/staggered.py:73: NameError

CC @AA-Turner

CPython versions tested on:

3.13, CPython main branch

Operating systems tested on:

Linux

Linked PRs

• gh-119121: Fix and test async.staggered.staggered_race #119173
• [3.13] gh-119121: Fix and test async.staggered.staggered_race (GH-119173) #119206

[Eclips4]

cc @sobolevn as well

[JelleZijlstra]

This is easy to fix but the weird thing is that no unit tests caught this; the NameError happens on any call to staggered_race(). We should really have test coverage for staggered.py.

Linking #114282 where this was changed.

[sobolevn]

I'll fix it, sorry!