feat: add unused-asyncio-task error code for asyncio.create_task without reference (#14710)

mypy/checker.py

@@ -96,7 +96,13 @@ def __init__(self) -> None:
     remove_instance_last_known_values,
     shallow_erase_type_for_equality,
 )
-from mypy.errorcodes import TYPE_VAR, UNUSED_AWAITABLE, UNUSED_COROUTINE, ErrorCode
+from mypy.errorcodes import (
+    TYPE_VAR,
+    UNUSED_ASYNCIO_TASK,
+    UNUSED_AWAITABLE,
+    UNUSED_COROUTINE,
+    ErrorCode,
+)
 from mypy.errors import (
     ErrorInfo,
     Errors,
@@ -5241,6 +5247,16 @@ def type_requires_usage(self, typ: Type, s: ExpressionStmt) -> tuple[str, ErrorC
         awaitables because in those particular cases we can guarantee
         it's safe.
         """
+        if isinstance(s.expr, CallExpr):
+            callee = s.expr.callee
+            if refers_to_fullname(callee, "asyncio.create_task") or refers_to_fullname(
+                callee, "asyncio.ensure_future"
+            ):
+                return (
+                    "The task may be garbage collected before it completes. "
+                    "Assign the result to a variable or add it to a task list.",
+                    UNUSED_ASYNCIO_TASK,
+                )
         proper_type = get_proper_type(typ)
         if isinstance(proper_type, Instance):
             # We use different error codes for generic awaitable vs coroutine.

mypy/errorcodes.py

@@ -238,6 +238,12 @@ def __hash__(self) -> int:
     "General",
     default_enabled=False,
 )
+UNUSED_ASYNCIO_TASK: Final = ErrorCode(
+    "unused-asyncio-task",
+    "Ensure that all asyncio tasks are used",
+    "General",
+    default_enabled=False,
+)
 REDUNDANT_SELF_TYPE: Final = ErrorCode(
     "redundant-self",
     "Warn about redundant Self type annotations on method first argument",

test-data/unit/check-async-await.test

@@ -896,6 +896,46 @@ class TaskGroup:
     async def __aenter__(self) -> "TaskGroup": pass
     async def __aexit__(self, x, y, z) -> None: pass
 
+[case testAsyncioCreateTaskWithoutReference]
+# flags: --enable-error-code unused-asyncio-task --show-error-codes
+import asyncio
+
+async def main() -> None:
+    asyncio.create_task(coro())  # E: Value of type "Any" must be used  [unused-asyncio-task] \
+                                 # N: The task may be garbage collected before it completes. Assign the result to a variable or add it to a task list.
+    asyncio.ensure_future(coro())  # E: Value of type "Any" must be used  [unused-asyncio-task] \
+                                   # N: The task may be garbage collected before it completes. Assign the result to a variable or add it to a task list.
+
+async def coro() -> None:
+    pass
+
+[file asyncio/__init__.pyi]
+from typing import Coroutine, Any
+def create_task(coro: Coroutine[Any, Any, Any]) -> Any: pass
+def ensure_future(coro: Coroutine[Any, Any, Any]) -> Any: pass
+[builtins fixtures/async_await.pyi]
+[typing fixtures/typing-async.pyi]
+
+[case testAsyncioCreateTaskWithReference]
+# flags: --enable-error-code unused-asyncio-task --show-error-codes
+import asyncio
+
+async def main() -> None:
+    task = asyncio.create_task(coro())
+    task2 = asyncio.ensure_future(coro())
+    await asyncio.create_task(coro())
+    await asyncio.ensure_future(coro())
+
+async def coro() -> None:
+    pass
+
+[file asyncio/__init__.pyi]
+from typing import Coroutine, Any
+def create_task(coro: Coroutine[Any, Any, Any]) -> Any: pass
+def ensure_future(coro: Coroutine[Any, Any, Any]) -> Any: pass
+[builtins fixtures/async_await.pyi]
+[typing fixtures/typing-async.pyi]
+
 [case testAsyncForOutsideCoroutine]
 async def g():
     yield 0

test-data/unit/check-overloading.test

@@ -6952,3 +6952,38 @@ reveal_type(f(la))  # N: Revealed type is "builtins.list[builtins.tuple[Any, ...
 reveal_type(f(li))  # N: Revealed type is "builtins.list[builtins.tuple[builtins.int, ...]]"
 reveal_type(f(ls))  # N: Revealed type is "builtins.list[builtins.tuple[Any, ...]]"
 [builtins fixtures/tuple.pyi]
+
+[case testOverloadOnlyWidenDisagreeingTypeParams]
+from collections.abc import Sequence
+from typing import Any, Generic, Protocol, assert_type, overload
+from typing_extensions import TypeVar
+
+S1 = TypeVar('S1')
+
+class EA: ...
+
+A = TypeVar('A', bound=EA, default=EA, covariant=True)
+S2 = TypeVar('S2', bound=EA)
+S2_contra = TypeVar('S2_contra', bound=EA, contravariant=True)
+
+class Supports_ProtoAdd(Protocol[S2_contra, S2]):
+    def _proto_add(self, other: S2_contra, /) -> 'Ser2[S2, EA]': ...
+
+class Ser2(Generic[S1, A]):
+    def _proto_add(self, other: Any, /) -> 'Ser2[Any, EA]':
+        pass
+
+    @overload
+    def __add__(
+        self: Supports_ProtoAdd[S2_contra, S2],
+        other: 'S2_contra | Sequence[S2_contra]',
+    ) -> 'Ser2[S2]': ...
+    @overload
+    def __add__(self: 'Ser2[bool, EA]', other: Sequence[Any]) -> 'Ser2[int]': ...
+    def __add__(self, other: Any) -> Any:
+        pass
+
+def f(a2: 'Ser2[Any, EA]', seq: 'Sequence[Any]') -> None:
+    reveal_type(a2 + seq)  # N: Revealed type is "__main__.Ser2[Any, __main__.EA]"
+    assert_type(a2 + seq, 'Ser2[Any, EA]')
+[builtins fixtures/tuple.pyi]