Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
8 changes: 4 additions & 4 deletions stdlib/@tests/test_cases/asyncio/check_coroutines.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,10 +19,10 @@ def test_iscoroutinefunction_asyncio(
assert_type(x, Callable[[str, int], Coroutine[str, int, bytes]])

if iscoroutinefunction(y): # pyright: ignore
assert_type(y, Callable[[str, int], Coroutine[Any, Any, bytes]])
assert_type(y, Callable[[str, int], Awaitable[bytes]])

if iscoroutinefunction(z): # pyright: ignore
assert_type(z, Callable[[str, int], Coroutine[Any, Any, Any]])
assert_type(z, Callable[[str, int], str | Awaitable[bytes]])

if iscoroutinefunction(xx): # pyright: ignore
assert_type(xx, Callable[..., Coroutine[Any, Any, Any]])
Expand All @@ -31,10 +31,10 @@ def test_iscoroutinefunction_asyncio(
assert_type(x, Callable[[str, int], Coroutine[str, int, bytes]])

if iscoroutinefunction(y):
assert_type(y, Callable[[str, int], Coroutine[Any, Any, bytes]])
assert_type(y, Callable[[str, int], Awaitable[bytes]])

if iscoroutinefunction(z):
assert_type(z, Callable[[str, int], Coroutine[Any, Any, Any]])
assert_type(z, Callable[[str, int], str | Awaitable[bytes]])

if iscoroutinefunction(xx):
assert_type(xx, Callable[..., Coroutine[Any, Any, Any]])
4 changes: 2 additions & 2 deletions stdlib/@tests/test_cases/check_inspect.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,10 @@ def test_iscoroutinefunction_inspect(
assert_type(x, Callable[[str, int], Coroutine[str, int, bytes]])

if inspect.iscoroutinefunction(y):
assert_type(y, Callable[[str, int], CoroutineType[Any, Any, bytes]])
assert_type(y, Callable[[str, int], Awaitable[bytes]])

if inspect.iscoroutinefunction(z):
assert_type(z, Callable[[str, int], CoroutineType[Any, Any, Any]])
assert_type(z, Callable[[str, int], str | Awaitable[bytes]])

if inspect.iscoroutinefunction(xx):
assert_type(xx, Callable[..., CoroutineType[Any, Any, Any]])
14 changes: 7 additions & 7 deletions stdlib/asyncio/coroutines.pyi
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import sys
from collections.abc import Awaitable, Callable, Coroutine
from typing import Any, ParamSpec, TypeGuard, TypeVar, overload
from typing import Any, ParamSpec, TypeVar, overload
from typing_extensions import TypeIs, deprecated

# Keep asyncio.__all__ updated with any changes to __all__ here
Expand All @@ -25,21 +25,21 @@ if sys.version_info >= (3, 11):
def iscoroutinefunction(func: Callable[..., Coroutine[Any, Any, Any]]) -> bool: ...
@overload
@deprecated("Deprecated since Python 3.14. Use `inspect.iscoroutinefunction()` instead.")
def iscoroutinefunction(func: Callable[_P, Awaitable[_T]]) -> TypeGuard[Callable[_P, Coroutine[Any, Any, _T]]]: ...
def iscoroutinefunction(func: Callable[_P, Awaitable[_T]]) -> TypeIs[Callable[_P, Coroutine[Any, Any, _T]]]: ...
@overload
@deprecated("Deprecated since Python 3.14. Use `inspect.iscoroutinefunction()` instead.")
def iscoroutinefunction(func: Callable[_P, object]) -> TypeGuard[Callable[_P, Coroutine[Any, Any, Any]]]: ...
def iscoroutinefunction(func: Callable[_P, object]) -> TypeIs[Callable[_P, Coroutine[Any, Any, Any]]]: ...
@overload
@deprecated("Deprecated since Python 3.14. Use `inspect.iscoroutinefunction()` instead.")
def iscoroutinefunction(func: object) -> TypeGuard[Callable[..., Coroutine[Any, Any, Any]]]: ...
def iscoroutinefunction(func: object) -> TypeIs[Callable[..., Coroutine[Any, Any, Any]]]: ...
else:
# Sometimes needed in Python < 3.11 due to the fact that it supports @coroutine
# which was removed in 3.11 which the inspect version doesn't support.
@overload
def iscoroutinefunction(func: Callable[..., Coroutine[Any, Any, Any]]) -> bool: ...
@overload
def iscoroutinefunction(func: Callable[_P, Awaitable[_T]]) -> TypeGuard[Callable[_P, Coroutine[Any, Any, _T]]]: ...
def iscoroutinefunction(func: Callable[_P, Awaitable[_T]]) -> TypeIs[Callable[_P, Coroutine[Any, Any, _T]]]: ...
@overload
def iscoroutinefunction(func: Callable[_P, object]) -> TypeGuard[Callable[_P, Coroutine[Any, Any, Any]]]: ...
def iscoroutinefunction(func: Callable[_P, object]) -> TypeIs[Callable[_P, Coroutine[Any, Any, Any]]]: ...
@overload
def iscoroutinefunction(func: object) -> TypeGuard[Callable[..., Coroutine[Any, Any, Any]]]: ...
def iscoroutinefunction(func: object) -> TypeIs[Callable[..., Coroutine[Any, Any, Any]]]: ...
6 changes: 3 additions & 3 deletions stdlib/inspect.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -239,11 +239,11 @@ def isgeneratorfunction(obj: object) -> TypeGuard[Callable[..., GeneratorType[An
@overload
def iscoroutinefunction(obj: Callable[..., Coroutine[Any, Any, Any]]) -> bool: ...
@overload
def iscoroutinefunction(obj: Callable[_P, Awaitable[_T]]) -> TypeGuard[Callable[_P, CoroutineType[Any, Any, _T]]]: ...
def iscoroutinefunction(obj: Callable[_P, Awaitable[_T]]) -> TypeIs[Callable[_P, CoroutineType[Any, Any, _T]]]: ...

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is incorrect because it does not return true for all callables that return a CoroutineType.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps a new type needs to be added to represent async functions, because with the current typing system there is no way to type

def foo(a) -> Awaitable[int]:
    return asyncio.Future()

as being distinct from

async def foo(a) -> int:
    return 5

and no way to distinguish them at runtime other than iscoroutinefunction, which can't be accurately typed as the annotations indicate a narrowing on the return type, while the logic of iscoroutinefunction is using flags on the function's code object.

Thus, you can't know definitively if a function is asynchronous in the current system.

@overload
def iscoroutinefunction(obj: Callable[_P, object]) -> TypeGuard[Callable[_P, CoroutineType[Any, Any, Any]]]: ...
def iscoroutinefunction(obj: Callable[_P, object]) -> TypeIs[Callable[_P, CoroutineType[Any, Any, Any]]]: ...
@overload
def iscoroutinefunction(obj: object) -> TypeGuard[Callable[..., CoroutineType[Any, Any, Any]]]: ...
def iscoroutinefunction(obj: object) -> TypeIs[Callable[..., CoroutineType[Any, Any, Any]]]: ...

def isgenerator(object: object) -> TypeIs[GeneratorType[Any, Any, Any]]: ...
def iscoroutine(object: object) -> TypeIs[CoroutineType[Any, Any, Any]]: ...
Expand Down
Loading