Skip to content

gh-128559: Remove typing import from asyncio.timeouts #128560

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Jan 6, 2025
Merged

gh-128559: Remove typing import from asyncio.timeouts #128560

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ef7e24f
Remove typing import from asyncio.timeouts
mikeshardmind Jan 6, 2025
0a64c83
📜🤖 Added by blurb_it.
blurb-it[bot] Jan 6, 2025
4ae5f5e
Update Misc/NEWS.d/next/Library/2025-01-06-21-35-00.gh-issue-128559.6…
mikeshardmind Jan 6, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 11 additions & 13 deletions Lib/asyncio/timeouts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import enum

from types import TracebackType
from typing import final, Optional, Type

from . import events
from . import exceptions
Expand All @@ -23,14 +22,13 @@ class _State(enum.Enum):
EXITED = "finished"


@final
class Timeout:
"""Asynchronous context manager for cancelling overdue coroutines.

Use `timeout()` or `timeout_at()` rather than instantiating this class directly.
"""

def __init__(self, when: Optional[float]) -> None:
def __init__(self, when: float | None) -> None:
"""Schedule a timeout that will trigger at a given loop time.

- If `when` is `None`, the timeout will never trigger.
Expand All @@ -39,15 +37,15 @@ def __init__(self, when: Optional[float]) -> None:
"""
self._state = _State.CREATED

self._timeout_handler: Optional[events.TimerHandle] = None
self._task: Optional[tasks.Task] = None
self._timeout_handler: events.TimerHandle | None = None
self._task: tasks.Task | None = None
self._when = when

def when(self) -> Optional[float]:
def when(self) -> float | None:
"""Return the current deadline."""
return self._when

def reschedule(self, when: Optional[float]) -> None:
def reschedule(self, when: float | None) -> None:
"""Reschedule the timeout."""
if self._state is not _State.ENTERED:
if self._state is _State.CREATED:
Expand Down Expand Up @@ -96,10 +94,10 @@ async def __aenter__(self) -> "Timeout":

async def __aexit__(
self,
exc_type: Optional[Type[BaseException]],
exc_val: Optional[BaseException],
exc_tb: Optional[TracebackType],
) -> Optional[bool]:
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> bool | None:
assert self._state in (_State.ENTERED, _State.EXPIRING)

if self._timeout_handler is not None:
Expand Down Expand Up @@ -142,7 +140,7 @@ def _insert_timeout_error(exc_val: BaseException) -> None:
exc_val = exc_val.__context__


def timeout(delay: Optional[float]) -> Timeout:
def timeout(delay: float | None) -> Timeout:
"""Timeout async context manager.

Useful in cases when you want to apply timeout logic around block
Expand All @@ -162,7 +160,7 @@ def timeout(delay: Optional[float]) -> Timeout:
return Timeout(loop.time() + delay if delay is not None else None)


def timeout_at(when: Optional[float]) -> Timeout:
def timeout_at(when: float | None) -> Timeout:
"""Schedule the timeout at absolute time.

Like timeout() but argument gives absolute time in the same clock system
Expand Down
1 change: 1 addition & 0 deletions Misc/NEWS.d/next/Library/2025-01-06-21-35-00.gh-issue-128559.6fxcDM.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Improved import time of :mod:`asyncio`.
Loading