Use TypeVarTuple for add_job and run_job methods (#114122)

This commit is contained in:
Marc Mueller
2024-03-24 18:52:39 +01:00
committed by GitHub
parent 67ab49b825
commit d779333bef
3 changed files with 23 additions and 17 deletions

View File

@@ -40,6 +40,7 @@ from typing import (
ParamSpec,
Self,
TypedDict,
TypeVarTuple,
cast,
overload,
)
@@ -135,6 +136,7 @@ _T = TypeVar("_T")
_R = TypeVar("_R")
_R_co = TypeVar("_R_co", covariant=True)
_P = ParamSpec("_P")
_Ts = TypeVarTuple("_Ts")
# Internal; not helpers.typing.UNDEFINED due to circular dependency
_UNDEF: dict[Any, Any] = {}
_CallableT = TypeVar("_CallableT", bound=Callable[..., Any])
@@ -529,7 +531,7 @@ class HomeAssistant:
self.bus.async_fire(EVENT_HOMEASSISTANT_STARTED)
def add_job(
self, target: Callable[..., Any] | Coroutine[Any, Any, Any], *args: Any
self, target: Callable[[*_Ts], Any] | Coroutine[Any, Any, Any], *args: *_Ts
) -> None:
"""Add a job to be executed by the event loop or by an executor.
@@ -547,7 +549,7 @@ class HomeAssistant:
)
return
if TYPE_CHECKING:
target = cast(Callable[..., Any], target)
target = cast(Callable[[*_Ts], Any], target)
self.loop.call_soon_threadsafe(
functools.partial(
self.async_add_hass_job, HassJob(target), *args, eager_start=True
@@ -558,8 +560,8 @@ class HomeAssistant:
@callback
def async_add_job(
self,
target: Callable[..., Coroutine[Any, Any, _R]],
*args: Any,
target: Callable[[*_Ts], Coroutine[Any, Any, _R]],
*args: *_Ts,
eager_start: bool = False,
) -> asyncio.Future[_R] | None:
...
@@ -568,8 +570,8 @@ class HomeAssistant:
@callback
def async_add_job(
self,
target: Callable[..., Coroutine[Any, Any, _R] | _R],
*args: Any,
target: Callable[[*_Ts], Coroutine[Any, Any, _R] | _R],
*args: *_Ts,
eager_start: bool = False,
) -> asyncio.Future[_R] | None:
...
@@ -587,8 +589,9 @@ class HomeAssistant:
@callback
def async_add_job(
self,
target: Callable[..., Coroutine[Any, Any, _R] | _R] | Coroutine[Any, Any, _R],
*args: Any,
target: Callable[[*_Ts], Coroutine[Any, Any, _R] | _R]
| Coroutine[Any, Any, _R],
*args: *_Ts,
eager_start: bool = False,
) -> asyncio.Future[_R] | None:
"""Add a job to be executed by the event loop or by an executor.
@@ -623,7 +626,7 @@ class HomeAssistant:
# the type used for the cast. For history see:
# https://github.com/home-assistant/core/pull/71960
if TYPE_CHECKING:
target = cast(Callable[..., Coroutine[Any, Any, _R] | _R], target)
target = cast(Callable[[*_Ts], Coroutine[Any, Any, _R] | _R], target)
return self.async_add_hass_job(HassJob(target), *args, eager_start=eager_start)
@overload
@@ -772,7 +775,7 @@ class HomeAssistant:
@callback
def async_add_executor_job(
self, target: Callable[..., _T], *args: Any
self, target: Callable[[*_Ts], _T], *args: *_Ts
) -> asyncio.Future[_T]:
"""Add an executor job from within the event loop."""
task = self.loop.run_in_executor(None, target, *args)
@@ -783,7 +786,7 @@ class HomeAssistant:
@callback
def async_add_import_executor_job(
self, target: Callable[..., _T], *args: Any
self, target: Callable[[*_Ts], _T], *args: *_Ts
) -> asyncio.Future[_T]:
"""Add an import executor job from within the event loop."""
task = self.loop.run_in_executor(self.import_executor, target, *args)
@@ -844,14 +847,14 @@ class HomeAssistant:
@overload
@callback
def async_run_job(
self, target: Callable[..., Coroutine[Any, Any, _R]], *args: Any
self, target: Callable[[*_Ts], Coroutine[Any, Any, _R]], *args: *_Ts
) -> asyncio.Future[_R] | None:
...
@overload
@callback
def async_run_job(
self, target: Callable[..., Coroutine[Any, Any, _R] | _R], *args: Any
self, target: Callable[[*_Ts], Coroutine[Any, Any, _R] | _R], *args: *_Ts
) -> asyncio.Future[_R] | None:
...
@@ -865,8 +868,9 @@ class HomeAssistant:
@callback
def async_run_job(
self,
target: Callable[..., Coroutine[Any, Any, _R] | _R] | Coroutine[Any, Any, _R],
*args: Any,
target: Callable[[*_Ts], Coroutine[Any, Any, _R] | _R]
| Coroutine[Any, Any, _R],
*args: *_Ts,
) -> asyncio.Future[_R] | None:
"""Run a job from within the event loop.
@@ -894,7 +898,7 @@ class HomeAssistant:
# the type used for the cast. For history see:
# https://github.com/home-assistant/core/pull/71960
if TYPE_CHECKING:
target = cast(Callable[..., Coroutine[Any, Any, _R] | _R], target)
target = cast(Callable[[*_Ts], Coroutine[Any, Any, _R] | _R], target)
return self.async_run_hass_job(HassJob(target), *args)
def block_till_done(self) -> None: