Movatterモバイル変換


[0]ホーム

URL:


Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

Sign up
Appearance settings

asyncio.staggered is missingtyping import #119121

Closed
Assignees
sobolevn
@mgorny

Description

@mgorny

Bug report

Bug description:

asyncio/staggered.py is usingtyping.Optional:

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

CC@AA-Turner

CPython versions tested on:

3.13, CPython main branch

Operating systems tested on:

Linux

Linked PRs

Metadata

Metadata

Assignees

Projects

Status

Done

Milestone

No milestone

Relationships

None yet

Development

No branches or pull requests

Issue actions


    [8]ページ先頭

    ©2009-2025 Movatter.jp