Home-Assistant
/
core
mirror of https://github.com/home-assistant/core.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Better type hass_job method calls (#76053)

pull/76531/head
Marc Mueller 2022-08-09 22:12:33 +02:00 committed by GitHub
parent 70aeaa3c76
commit dc47121f2c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 37 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
homeassistant/core.py
View File

@ -816,7 +816,7 @@ class Event:
class _FilterableJob(NamedTuple):
"""Event listener job to be executed with optional filter."""
job: HassJob[[Event], None | Awaitable[None]]
job: HassJob[[Event], Coroutine[Any, Any, None] | None]
event_filter: Callable[[Event], bool] | None
run_immediately: bool
@ -907,7 +907,7 @@ class EventBus:
def listen(
self,
event_type: str,
listener: Callable[[Event], None | Awaitable[None]],
listener: Callable[[Event], Coroutine[Any, Any, None] | None],
) -> CALLBACK_TYPE:
"""Listen for all events or events of a specific type.
@ -928,7 +928,7 @@ class EventBus:
def async_listen(
self,
event_type: str,
listener: Callable[[Event], None | Awaitable[None]],
listener: Callable[[Event], Coroutine[Any, Any, None] | None],
event_filter: Callable[[Event], bool] | None = None,
run_immediately: bool = False,
) -> CALLBACK_TYPE:
@ -968,7 +968,9 @@ class EventBus:
return remove_listener
def listen_once(
self, event_type: str, listener: Callable[[Event], None | Awaitable[None]]
self,
event_type: str,
listener: Callable[[Event], Coroutine[Any, Any, None] | None],
) -> CALLBACK_TYPE:
"""Listen once for event of a specific type.
@ -989,7 +991,9 @@ class EventBus:
@callback
def async_listen_once(
self, event_type: str, listener: Callable[[Event], None | Awaitable[None]]
self,
event_type: str,
listener: Callable[[Event], Coroutine[Any, Any, None] | None],
) -> CALLBACK_TYPE:
"""Listen once for event of a specific type.
@ -1463,7 +1467,7 @@ class Service:
def __init__(
self,
func: Callable[[ServiceCall], None | Awaitable[None]],
func: Callable[[ServiceCall], Coroutine[Any, Any, None] | None],
schema: vol.Schema | None,
context: Context | None = None,
) -> None:
@ -1533,7 +1537,7 @@ class ServiceRegistry:
self,
domain: str,
service: str,
service_func: Callable[[ServiceCall], Awaitable[None] | None],
service_func: Callable[[ServiceCall], Coroutine[Any, Any, None] | None],
schema: vol.Schema | None = None,
) -> None:
"""
@ -1550,7 +1554,7 @@ class ServiceRegistry:
self,
domain: str,
service: str,
service_func: Callable[[ServiceCall], Awaitable[None] | None],
service_func: Callable[[ServiceCall], Coroutine[Any, Any, None] | None],
schema: vol.Schema | None = None,
) -> None:
"""

6
homeassistant/helpers/discovery.py
View File

@ -7,7 +7,7 @@ There are two different types of discoveries that can be fired/listened for.
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from collections.abc import Callable, Coroutine
from typing import Any, TypedDict
from homeassistant import core, setup
@ -36,7 +36,9 @@ class DiscoveryDict(TypedDict):
def async_listen(
hass: core.HomeAssistant,
service: str,
callback: Callable[[str, DiscoveryInfoType | None], Awaitable[None] | None],
callback: Callable[
[str, DiscoveryInfoType | None], Coroutine[Any, Any, None] | None
],
) -> None:
"""Set up listener for discovery of specific service.

32
homeassistant/helpers/event.py
View File

@ -2,7 +2,7 @@
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable, Iterable, Sequence
from collections.abc import Callable, Coroutine, Iterable, Sequence
import copy
from dataclasses import dataclass
from datetime import datetime, timedelta
@ -141,7 +141,7 @@ def threaded_listener_factory(
def async_track_state_change(
hass: HomeAssistant,
entity_ids: str | Iterable[str],
action: Callable[[str, State | None, State], Awaitable[None] | None],
action: Callable[[str, State | None, State], Coroutine[Any, Any, None] | None],
from_state: None | str | Iterable[str] = None,
to_state: None | str | Iterable[str] = None,
) -> CALLBACK_TYPE:
@ -714,7 +714,9 @@ def async_track_state_change_filtered(
def async_track_template(
hass: HomeAssistant,
template: Template,
action: Callable[[str, State | None, State | None], Awaitable[None] | None],
action: Callable[
[str, State | None, State | None], Coroutine[Any, Any, None] | None
],
variables: TemplateVarsType | None = None,
) -> CALLBACK_TYPE:
"""Add a listener that fires when a a template evaluates to 'true'.
@ -1188,7 +1190,7 @@ def async_track_template_result(
def async_track_same_state(
hass: HomeAssistant,
period: timedelta,
action: Callable[[], Awaitable[None] | None],
action: Callable[[], Coroutine[Any, Any, None] | None],
async_check_same_func: Callable[[str, State | None, State | None], bool],
entity_ids: str | Iterable[str] = MATCH_ALL,
) -> CALLBACK_TYPE:
@ -1257,8 +1259,8 @@ track_same_state = threaded_listener_factory(async_track_same_state)
@bind_hass
def async_track_point_in_time(
hass: HomeAssistant,
action: HassJob[[datetime], Awaitable[None] | None]
| Callable[[datetime], Awaitable[None] | None],
action: HassJob[[datetime], Coroutine[Any, Any, None] | None]
| Callable[[datetime], Coroutine[Any, Any, None] | None],
point_in_time: datetime,
) -> CALLBACK_TYPE:
"""Add a listener that fires once after a specific point in time."""
@ -1279,8 +1281,8 @@ track_point_in_time = threaded_listener_factory(async_track_point_in_time)
@bind_hass
def async_track_point_in_utc_time(
hass: HomeAssistant,
action: HassJob[[datetime], Awaitable[None] | None]
| Callable[[datetime], Awaitable[None] | None],
action: HassJob[[datetime], Coroutine[Any, Any, None] | None]
| Callable[[datetime], Coroutine[Any, Any, None] | None],
point_in_time: datetime,
) -> CALLBACK_TYPE:
"""Add a listener that fires once after a specific point in UTC time."""
@ -1293,7 +1295,7 @@ def async_track_point_in_utc_time(
cancel_callback: asyncio.TimerHandle | None = None
@callback
def run_action(job: HassJob[[datetime], Awaitable[None] | None]) -> None:
def run_action(job: HassJob[[datetime], Coroutine[Any, Any, None] | None]) -> None:
"""Call the action."""
nonlocal cancel_callback
# Depending on the available clock support (including timer hardware
@ -1330,8 +1332,8 @@ track_point_in_utc_time = threaded_listener_factory(async_track_point_in_utc_tim
def async_call_later(
hass: HomeAssistant,
delay: float | timedelta,
action: HassJob[[datetime], Awaitable[None] | None]
| Callable[[datetime], Awaitable[None] | None],
action: HassJob[[datetime], Coroutine[Any, Any, None] | None]
| Callable[[datetime], Coroutine[Any, Any, None] | None],
) -> CALLBACK_TYPE:
"""Add a listener that is called in <delay>."""
if not isinstance(delay, timedelta):
@ -1346,7 +1348,7 @@ call_later = threaded_listener_factory(async_call_later)
@bind_hass
def async_track_time_interval(
hass: HomeAssistant,
action: Callable[[datetime], Awaitable[None] | None],
action: Callable[[datetime], Coroutine[Any, Any, None] | None],
interval: timedelta,
) -> CALLBACK_TYPE:
"""Add a listener that fires repetitively at every timedelta interval."""
@ -1388,7 +1390,7 @@ class SunListener:
"""Helper class to help listen to sun events."""
hass: HomeAssistant = attr.ib()
job: HassJob[[], Awaitable[None] | None] = attr.ib()
job: HassJob[[], Coroutine[Any, Any, None] | None] = attr.ib()
event: str = attr.ib()
offset: timedelta | None = attr.ib()
_unsub_sun: CALLBACK_TYPE | None = attr.ib(default=None)
@ -1479,7 +1481,7 @@ time_tracker_timestamp = time.time
@bind_hass
def async_track_utc_time_change(
hass: HomeAssistant,
action: Callable[[datetime], Awaitable[None] | None],
action: Callable[[datetime], Coroutine[Any, Any, None] | None],
hour: Any | None = None,
minute: Any | None = None,
second: Any | None = None,
@ -1544,7 +1546,7 @@ track_utc_time_change = threaded_listener_factory(async_track_utc_time_change)
@bind_hass
def async_track_time_change(
hass: HomeAssistant,
action: Callable[[datetime], Awaitable[None] | None],
action: Callable[[datetime], Coroutine[Any, Any, None] | None],
hour: Any | None = None,
minute: Any | None = None,
second: Any | None = None,

6
homeassistant/helpers/start.py
View File

@ -1,7 +1,8 @@
"""Helpers to help during startup."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from collections.abc import Callable, Coroutine
from typing import Any
from homeassistant.const import EVENT_HOMEASSISTANT_START
from homeassistant.core import CALLBACK_TYPE, Event, HassJob, HomeAssistant, callback
@ -9,7 +10,8 @@ from homeassistant.core import CALLBACK_TYPE, Event, HassJob, HomeAssistant, cal
@callback
def async_at_start(
hass: HomeAssistant, at_start_cb: Callable[[HomeAssistant], Awaitable[None] | None]
hass: HomeAssistant,
at_start_cb: Callable[[HomeAssistant], Coroutine[Any, Any, None] | None],
) -> CALLBACK_TYPE:
"""Execute something when Home Assistant is started.