Use _AwaitableLike in EventLoop.run_until_complete (#10181)

This commit is contained in:
Nikita Sobolev
2023-05-15 11:34:50 +03:00
committed by GitHub
parent 3032e88595
commit b25a5b6aef
3 changed files with 9 additions and 16 deletions

View File

@@ -1,5 +1,5 @@
import sys
from collections.abc import Coroutine, Generator
from collections.abc import Awaitable, Coroutine, Generator
from typing import Any, TypeVar
from typing_extensions import TypeAlias
@@ -36,6 +36,8 @@ _T = TypeVar("_T")
# Aliases imported by multiple submodules in typeshed
if sys.version_info >= (3, 12):
_AwaitableLike: TypeAlias = Awaitable[_T]
_CoroutineLike: TypeAlias = Coroutine[Any, Any, _T]
else:
_AwaitableLike: TypeAlias = Generator[Any, None, _T] | Awaitable[_T]
_CoroutineLike: TypeAlias = Generator[Any, None, _T] | Coroutine[Any, Any, _T]

View File

@@ -1,13 +1,13 @@
import ssl
import sys
from _typeshed import FileDescriptorLike, ReadableBuffer, WriteableBuffer
from asyncio import _CoroutineLike
from asyncio import _AwaitableLike, _CoroutineLike
from asyncio.events import AbstractEventLoop, AbstractServer, Handle, TimerHandle, _TaskFactory
from asyncio.futures import Future
from asyncio.protocols import BaseProtocol
from asyncio.tasks import Task
from asyncio.transports import BaseTransport, DatagramTransport, ReadTransport, SubprocessTransport, Transport, WriteTransport
from collections.abc import Awaitable, Callable, Generator, Iterable, Sequence
from collections.abc import Callable, Iterable, Sequence
from contextvars import Context
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
from typing import IO, Any, TypeVar, overload
@@ -64,11 +64,7 @@ class Server(AbstractServer):
class BaseEventLoop(AbstractEventLoop):
def run_forever(self) -> None: ...
# Can't use a union, see mypy issue # 1873.
@overload
def run_until_complete(self, future: Generator[Any, None, _T]) -> _T: ...
@overload
def run_until_complete(self, future: Awaitable[_T]) -> _T: ...
def run_until_complete(self, future: _AwaitableLike[_T]) -> _T: ...
def stop(self) -> None: ...
def is_running(self) -> bool: ...
def is_closed(self) -> bool: ...

View File

@@ -2,13 +2,13 @@ import ssl
import sys
from _typeshed import FileDescriptorLike, ReadableBuffer, StrPath, Unused, WriteableBuffer
from abc import ABCMeta, abstractmethod
from collections.abc import Awaitable, Callable, Coroutine, Generator, Sequence
from collections.abc import Callable, Coroutine, Generator, Sequence
from contextvars import Context
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
from typing import IO, Any, Protocol, TypeVar, overload
from typing_extensions import Literal, Self, TypeAlias
from . import _CoroutineLike
from . import _AwaitableLike, _CoroutineLike
from .base_events import Server
from .futures import Future
from .protocols import BaseProtocol
@@ -113,13 +113,8 @@ class AbstractEventLoop:
slow_callback_duration: float
@abstractmethod
def run_forever(self) -> None: ...
# Can't use a union, see mypy issue # 1873.
@overload
@abstractmethod
def run_until_complete(self, future: Generator[Any, None, _T]) -> _T: ...
@overload
@abstractmethod
def run_until_complete(self, future: Awaitable[_T]) -> _T: ...
def run_until_complete(self, future: _AwaitableLike[_T]) -> _T: ...
@abstractmethod
def stop(self) -> None: ...
@abstractmethod