jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-17 02:07:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve typing for core add_job and run_job methods (#70702)

Browse Source
This commit is contained in:
Marc Mueller 2022-04-25 23:50:01 +02:00 committed by GitHub
parent 0d9191c344
commit c1d2017988
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 21 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
homeassistant/config_entries.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
import asyncio import asyncio
from collections import ChainMap from collections import ChainMap
from collections.abc import Awaitable, Callable, Iterable, Mapping from collections.abc import Callable, Coroutine, Iterable, Mapping
from contextvars import ContextVar from contextvars import ContextVar
import dataclasses import dataclasses
from enum import Enum from enum import Enum
@ -160,7 +160,7 @@ class OperationNotAllowed(ConfigError):
"""Raised when a config entry operation is not allowed.""" """Raised when a config entry operation is not allowed."""
UpdateListenerType = Callable[[HomeAssistant, "ConfigEntry"], Awaitable[None]] UpdateListenerType = Callable[[HomeAssistant, "ConfigEntry"], Coroutine[Any, Any, None]]
class ConfigEntry: class ConfigEntry:

37
homeassistant/core.py
View File

@ -31,6 +31,7 @@ from typing import (
NamedTuple, NamedTuple,
Optional, Optional,
TypeVar, TypeVar,
Union,
cast, cast,
overload, overload,
) )
@ -363,14 +364,14 @@ class HomeAssistant:
@overload @overload
@callback @callback
def async_add_job( def async_add_job(
self, target: Callable[..., Awaitable[_R]], *args: Any self, target: Callable[..., Coroutine[Any, Any, _R]], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
... ...
@overload @overload
@callback @callback
def async_add_job( def async_add_job(
self, target: Callable[..., Awaitable[_R] | _R], *args: Any self, target: Callable[..., Coroutine[Any, Any, _R] | _R], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
... ...
@ -384,7 +385,7 @@ class HomeAssistant:
@callback @callback
def async_add_job( def async_add_job(
self, self,
target: Callable[..., Awaitable[_R] | _R] | Coroutine[Any, Any, _R], target: Callable[..., Coroutine[Any, Any, _R] | _R] | Coroutine[Any, Any, _R],
*args: Any, *args: Any,
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
"""Add a job to be executed by the event loop or by an executor. """Add a job to be executed by the event loop or by an executor.
@ -403,26 +404,26 @@ class HomeAssistant:
if asyncio.iscoroutine(target): if asyncio.iscoroutine(target):
return self.async_create_task(target) return self.async_create_task(target)
target = cast(Callable[..., _R], target) target = cast(Callable[..., Union[Coroutine[Any, Any, _R], _R]], target)
return self.async_add_hass_job(HassJob(target), *args) return self.async_add_hass_job(HassJob(target), *args)
@overload @overload
@callback @callback
def async_add_hass_job( def async_add_hass_job(
self, hassjob: HassJob[Awaitable[_R]], *args: Any self, hassjob: HassJob[Coroutine[Any, Any, _R]], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
... ...
@overload @overload
@callback @callback
def async_add_hass_job( def async_add_hass_job(
self, hassjob: HassJob[Awaitable[_R] | _R], *args: Any self, hassjob: HassJob[Coroutine[Any, Any, _R] | _R], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
... ...
@callback @callback
def async_add_hass_job( def async_add_hass_job(
self, hassjob: HassJob[Awaitable[_R] | _R], *args: Any self, hassjob: HassJob[Coroutine[Any, Any, _R] | _R], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
"""Add a HassJob from within the event loop. """Add a HassJob from within the event loop.
@ -433,10 +434,10 @@ class HomeAssistant:
task: asyncio.Future[_R] task: asyncio.Future[_R]
if hassjob.job_type == HassJobType.Coroutinefunction: if hassjob.job_type == HassJobType.Coroutinefunction:
task = self.loop.create_task( task = self.loop.create_task(
cast(Callable[..., Awaitable[_R]], hassjob.target)(*args) cast(Callable[..., Coroutine[Any, Any, _R]], hassjob.target)(*args)
) )
elif hassjob.job_type == HassJobType.Callback: elif hassjob.job_type == HassJobType.Callback:
self.loop.call_soon(hassjob.target, *args) self.loop.call_soon(cast(Callable[..., _R], hassjob.target), *args)
return None return None
else: else:
task = self.loop.run_in_executor( task = self.loop.run_in_executor(
@ -449,7 +450,7 @@ class HomeAssistant:
return task return task
def create_task(self, target: Awaitable[Any]) -> None: def create_task(self, target: Coroutine[Any, Any, Any]) -> None:
"""Add task to the executor pool. """Add task to the executor pool.
target: target to call. target: target to call.
@ -457,7 +458,7 @@ class HomeAssistant:
self.loop.call_soon_threadsafe(self.async_create_task, target) self.loop.call_soon_threadsafe(self.async_create_task, target)
@callback @callback
def async_create_task(self, target: Awaitable[_R]) -> asyncio.Task[_R]: def async_create_task(self, target: Coroutine[Any, Any, _R]) -> asyncio.Task[_R]:
"""Create a task from within the eventloop. """Create a task from within the eventloop.
This method must be run in the event loop. This method must be run in the event loop.
@ -497,20 +498,20 @@ class HomeAssistant:
@overload @overload
@callback @callback
def async_run_hass_job( def async_run_hass_job(
self, hassjob: HassJob[Awaitable[_R]], *args: Any self, hassjob: HassJob[Coroutine[Any, Any, _R]], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
... ...
@overload @overload
@callback @callback
def async_run_hass_job( def async_run_hass_job(
self, hassjob: HassJob[Awaitable[_R] | _R], *args: Any self, hassjob: HassJob[Coroutine[Any, Any, _R] | _R], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
... ...
@callback @callback
def async_run_hass_job( def async_run_hass_job(
self, hassjob: HassJob[Awaitable[_R] | _R], *args: Any self, hassjob: HassJob[Coroutine[Any, Any, _R] | _R], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
"""Run a HassJob from within the event loop. """Run a HassJob from within the event loop.
@ -528,14 +529,14 @@ class HomeAssistant:
@overload @overload
@callback @callback
def async_run_job( def async_run_job(
self, target: Callable[..., Awaitable[_R]], *args: Any self, target: Callable[..., Coroutine[Any, Any, _R]], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
... ...
@overload @overload
@callback @callback
def async_run_job( def async_run_job(
self, target: Callable[..., Awaitable[_R] | _R], *args: Any self, target: Callable[..., Coroutine[Any, Any, _R] | _R], *args: Any
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
... ...
@ -549,7 +550,7 @@ class HomeAssistant:
@callback @callback
def async_run_job( def async_run_job(
self, self,
target: Callable[..., Awaitable[_R] | _R] | Coroutine[Any, Any, _R], target: Callable[..., Coroutine[Any, Any, _R] | _R] | Coroutine[Any, Any, _R],
*args: Any, *args: Any,
) -> asyncio.Future[_R] | None: ) -> asyncio.Future[_R] | None:
"""Run a job from within the event loop. """Run a job from within the event loop.
@ -562,7 +563,7 @@ class HomeAssistant:
if asyncio.iscoroutine(target): if asyncio.iscoroutine(target):
return self.async_create_task(target) return self.async_create_task(target)
target = cast(Callable[..., _R], target) target = cast(Callable[..., Union[Coroutine[Any, Any, _R], _R]], target)
return self.async_run_hass_job(HassJob(target), *args) return self.async_run_hass_job(HassJob(target), *args)
def block_till_done(self) -> None: def block_till_done(self) -> None: