VimPlug/typeshed
1
0
Fork 0
mirror of https://github.com/davidhalter/typeshed.git synced 2025-12-22 20:01:29 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Require coroutines for create_task() (#6779)

Browse Source
This commit is contained in:
Nikita Sobolev
2022-02-02 15:35:00 +03:00
committed by GitHub
parent 970b8a676c
commit 1b99812621
3 changed files with 11 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
stdlib/asyncio/base_events.pyi
View File

@@ -9,7 +9,7 @@ from asyncio.tasks import Task
from asyncio.transports import BaseTransport, ReadTransport, SubprocessTransport, WriteTransport
from collections.abc import Iterable
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
from typing import IO, Any, Awaitable, Callable, Generator, Sequence, TypeVar, Union, overload
from typing import IO, Any, Awaitable, Callable, Coroutine, Generator, Sequence, TypeVar, Union, overload
from typing_extensions import Literal
if sys.version_info >= (3, 7):
@@ -75,9 +75,9 @@ class BaseEventLoop(AbstractEventLoop, metaclass=ABCMeta):
def create_future(self) -> Future[Any]: ...
# Tasks methods
if sys.version_info >= (3, 8):
def create_task(self, coro: Awaitable[_T] | Generator[Any, None, _T], *, name: object = ...) -> Task[_T]: ...
def create_task(self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T], *, name: object = ...) -> Task[_T]: ...
else:
def create_task(self, coro: Awaitable[_T] | Generator[Any, None, _T]) -> Task[_T]: ...
def create_task(self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T]) -> Task[_T]: ...
def set_task_factory(self, factory: Callable[[AbstractEventLoop, Generator[Any, None, _T]], Future[_T]] | None) -> None: ...
def get_task_factory(self) -> Callable[[AbstractEventLoop, Generator[Any, None, _T]], Future[_T]] | None: ...

8
stdlib/asyncio/events.pyi
View File

@@ -3,7 +3,7 @@ import sys
from _typeshed import FileDescriptorLike, Self
from abc import ABCMeta, abstractmethod
from socket import AddressFamily, SocketKind, _Address, _RetAddress, socket
from typing import IO, Any, Awaitable, Callable, Generator, Sequence, TypeVar, Union, overload
from typing import IO, Any, Awaitable, Callable, Coroutine, Generator, Sequence, TypeVar, Union, overload
from typing_extensions import Literal
from .base_events import Server
@@ -103,10 +103,12 @@ class AbstractEventLoop(metaclass=ABCMeta):
# Tasks methods
if sys.version_info >= (3, 8):
@abstractmethod
def create_task(self, coro: Awaitable[_T] | Generator[Any, None, _T], *, name: str | None = ...) -> Task[_T]: ...
def create_task(
self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T], *, name: str | None = ...
) -> Task[_T]: ...
else:
@abstractmethod
def create_task(self, coro: Awaitable[_T] | Generator[Any, None, _T]) -> Task[_T]: ...
def create_task(self, coro: Coroutine[Any, Any, _T] | Generator[Any, None, _T]) -> Task[_T]: ...
@abstractmethod
def set_task_factory(self, factory: Callable[[AbstractEventLoop, Generator[Any, None, _T]], Future[_T]] | None) -> None: ...

6
stdlib/asyncio/tasks.pyi
View File

@@ -2,7 +2,7 @@ import concurrent.futures
import sys
from collections.abc import Awaitable, Generator, Iterable, Iterator
from types import FrameType
from typing import Any, Generic, Optional, TextIO, TypeVar, Union, overload
from typing import Any, Coroutine, Generic, Optional, TextIO, TypeVar, Union, overload
from typing_extensions import Literal
from .events import AbstractEventLoop
@@ -289,9 +289,9 @@ class Task(Future[_T], Generic[_T]):
if sys.version_info >= (3, 7):
def all_tasks(loop: AbstractEventLoop | None = ...) -> set[Task[Any]]: ...
if sys.version_info >= (3, 8):
def create_task(coro: Generator[_TaskYieldType, None, _T] | Awaitable[_T], *, name: str | None = ...) -> Task[_T]: ...
def create_task(coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T], *, name: str | None = ...) -> Task[_T]: ...
else:
def create_task(coro: Generator[_TaskYieldType, None, _T] | Awaitable[_T]) -> Task[_T]: ...
def create_task(coro: Generator[Any, None, _T] | Coroutine[Any, Any, _T]) -> Task[_T]: ...
def current_task(loop: AbstractEventLoop | None = ...) -> Task[Any] | None: ...
def _enter_task(loop: AbstractEventLoop, task: Task[Any]) -> None: ...