Add first TypeVarTuple annotations (#105379)

homeassistant/helpers/ratelimit.py

@@ -5,11 +5,13 @@ import asyncio
 from collections.abc import Callable, Hashable
 from datetime import datetime, timedelta
 import logging
-from typing import Any
+from typing import TypeVarTuple
 
 from homeassistant.core import HomeAssistant, callback
 import homeassistant.util.dt as dt_util
 
+_Ts = TypeVarTuple("_Ts")
+
 _LOGGER = logging.getLogger(__name__)
 
 
@@ -59,8 +61,8 @@ class KeyedRateLimit:
         key: Hashable,
         rate_limit: timedelta | None,
         now: datetime,
-        action: Callable,
-        *args: Any,
+        action: Callable[[*_Ts], None],
+        *args: *_Ts,
     ) -> datetime | None:
         """Check rate limits and schedule an action if we hit the limit.
 

homeassistant/util/async_.py

@@ -10,7 +10,7 @@ import functools
 import logging
 import threading
 from traceback import extract_stack
-from typing import Any, ParamSpec, TypeVar
+from typing import Any, ParamSpec, TypeVar, TypeVarTuple
 
 from homeassistant.exceptions import HomeAssistantError
 
@@ -21,6 +21,7 @@ _SHUTDOWN_RUN_CALLBACK_THREADSAFE = "_shutdown_run_callback_threadsafe"
 _T = TypeVar("_T")
 _R = TypeVar("_R")
 _P = ParamSpec("_P")
+_Ts = TypeVarTuple("_Ts")
 
 
 def cancelling(task: Future[Any]) -> bool:
@@ -29,7 +30,7 @@ def cancelling(task: Future[Any]) -> bool:
 
 
 def run_callback_threadsafe(
-    loop: AbstractEventLoop, callback: Callable[..., _T], *args: Any
+    loop: AbstractEventLoop, callback: Callable[[*_Ts], _T], *args: *_Ts
 ) -> concurrent.futures.Future[_T]:
     """Submit a callback object to a given event loop.