Improve logging util typing (#105968)

homeassistant/util/logging.py

@@ -9,11 +9,12 @@ import logging
 import logging.handlers
 import queue
 import traceback
-from typing import Any, TypeVar, cast, overload
+from typing import Any, TypeVar, TypeVarTuple, cast, overload
 
 from homeassistant.core import HomeAssistant, callback, is_callback
 
 _T = TypeVar("_T")
+_Ts = TypeVarTuple("_Ts")
 
 
 class HomeAssistantQueueHandler(logging.handlers.QueueHandler):
@@ -83,7 +84,7 @@ def async_activate_log_queue_handler(hass: HomeAssistant) -> None:
     listener.start()
 
 
-def log_exception(format_err: Callable[..., Any], *args: Any) -> None:
+def log_exception(format_err: Callable[[*_Ts], Any], *args: *_Ts) -> None:
     """Log an exception with additional context."""
     module = inspect.getmodule(inspect.stack(context=0)[1].frame)
     if module is not None:
@@ -102,9 +103,9 @@ def log_exception(format_err: Callable[..., Any], *args: Any) -> None:
 
 
 async def _async_wrapper(
-    async_func: Callable[..., Coroutine[Any, Any, None]],
+    async_func: Callable[[*_Ts], Coroutine[Any, Any, None]],
-    format_err: Callable[..., Any],
+    format_err: Callable[[*_Ts], Any],
-    *args: Any,
+    *args: *_Ts,
 ) -> None:
     """Catch and log exception."""
     try:
@@ -114,7 +115,7 @@ async def _async_wrapper(
 
 
 def _sync_wrapper(
-    func: Callable[..., Any], format_err: Callable[..., Any], *args: Any
+    func: Callable[[*_Ts], Any], format_err: Callable[[*_Ts], Any], *args: *_Ts
 ) -> None:
     """Catch and log exception."""
     try:
@@ -125,7 +126,7 @@ def _sync_wrapper(
 
 @callback
 def _callback_wrapper(
-    func: Callable[..., Any], format_err: Callable[..., Any], *args: Any
+    func: Callable[[*_Ts], Any], format_err: Callable[[*_Ts], Any], *args: *_Ts
 ) -> None:
     """Catch and log exception."""
     try:
@@ -136,21 +137,21 @@ def _callback_wrapper(
 
 @overload
 def catch_log_exception(
-    func: Callable[..., Coroutine[Any, Any, Any]], format_err: Callable[..., Any]
+    func: Callable[[*_Ts], Coroutine[Any, Any, Any]], format_err: Callable[[*_Ts], Any]
-) -> Callable[..., Coroutine[Any, Any, None]]:
+) -> Callable[[*_Ts], Coroutine[Any, Any, None]]:
     ...
 
 
 @overload
 def catch_log_exception(
-    func: Callable[..., Any], format_err: Callable[..., Any]
+    func: Callable[[*_Ts], Any], format_err: Callable[[*_Ts], Any]
-) -> Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]:
+) -> Callable[[*_Ts], None] | Callable[[*_Ts], Coroutine[Any, Any, None]]:
     ...
 
 
 def catch_log_exception(
-    func: Callable[..., Any], format_err: Callable[..., Any]
+    func: Callable[[*_Ts], Any], format_err: Callable[[*_Ts], Any]
-) -> Callable[..., None] | Callable[..., Coroutine[Any, Any, None]]:
+) -> Callable[[*_Ts], None] | Callable[[*_Ts], Coroutine[Any, Any, None]]:
     """Decorate a function func to catch and log exceptions.
 
     If func is a coroutine function, a coroutine function will be returned.
@@ -159,24 +160,24 @@ def catch_log_exception(
     # Check for partials to properly determine if coroutine function
     check_func = func
     while isinstance(check_func, partial):
-        check_func = check_func.func
+        check_func = check_func.func  # type: ignore[unreachable]  # false positive
 
     if asyncio.iscoroutinefunction(check_func):
-        async_func = cast(Callable[..., Coroutine[Any, Any, None]], func)
+        async_func = cast(Callable[[*_Ts], Coroutine[Any, Any, None]], func)
-        return wraps(async_func)(partial(_async_wrapper, async_func, format_err))
+        return wraps(async_func)(partial(_async_wrapper, async_func, format_err))  # type: ignore[return-value]
 
     if is_callback(check_func):
-        return wraps(func)(partial(_callback_wrapper, func, format_err))
+        return wraps(func)(partial(_callback_wrapper, func, format_err))  # type: ignore[return-value]
 
-    return wraps(func)(partial(_sync_wrapper, func, format_err))
+    return wraps(func)(partial(_sync_wrapper, func, format_err))  # type: ignore[return-value]
 
 
 def catch_log_coro_exception(
-    target: Coroutine[Any, Any, _T], format_err: Callable[..., Any], *args: Any
+    target: Coroutine[Any, Any, _T], format_err: Callable[[*_Ts], Any], *args: *_Ts
 ) -> Coroutine[Any, Any, _T | None]:
     """Decorate a coroutine to catch and log exceptions."""
 
-    async def coro_wrapper(*args: Any) -> _T | None:
+    async def coro_wrapper(*args: *_Ts) -> _T | None:
         """Catch and log exception."""
         try:
             return await target