Use Callback protocol for AutomationActionType (#76054)

This commit is contained in:
Marc Mueller 2022-08-09 22:10:26 +02:00 committed by GitHub
parent ad361b8fc2
commit 70aeaa3c76
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 32 additions and 11 deletions

View File

@ -1,9 +1,9 @@
"""Allow to set up simple automation rules via the config file.""" """Allow to set up simple automation rules via the config file."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable from collections.abc import Callable
import logging import logging
from typing import Any, TypedDict, cast from typing import Any, Protocol, TypedDict, cast
import voluptuous as vol import voluptuous as vol
from voluptuous.humanize import humanize_error from voluptuous.humanize import humanize_error
@ -110,7 +110,17 @@ ATTR_VARIABLES = "variables"
SERVICE_TRIGGER = "trigger" SERVICE_TRIGGER = "trigger"
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
AutomationActionType = Callable[[HomeAssistant, TemplateVarsType], Awaitable[None]]
class AutomationActionType(Protocol):
"""Protocol type for automation action callback."""
async def __call__(
self,
run_variables: dict[str, Any],
context: Context | None = None,
) -> None:
"""Define action callback type."""
class AutomationTriggerData(TypedDict): class AutomationTriggerData(TypedDict):
@ -437,7 +447,12 @@ class AutomationEntity(ToggleEntity, RestoreEntity):
else: else:
await self.async_disable() await self.async_disable()
async def async_trigger(self, run_variables, context=None, skip_condition=False): async def async_trigger(
self,
run_variables: dict[str, Any],
context: Context | None = None,
skip_condition: bool = False,
) -> None:
"""Trigger automation. """Trigger automation.
This method is a coroutine. This method is a coroutine.
@ -462,7 +477,7 @@ class AutomationEntity(ToggleEntity, RestoreEntity):
this = None this = None
if state := self.hass.states.get(self.entity_id): if state := self.hass.states.get(self.entity_id):
this = state.as_dict() this = state.as_dict()
variables = {"this": this, **(run_variables or {})} variables: dict[str, Any] = {"this": this, **(run_variables or {})}
if self._variables: if self._variables:
try: try:
variables = self._variables.async_render(self.hass, variables) variables = self._variables.async_render(self.hass, variables)

View File

@ -1,6 +1,7 @@
"""Offer calendar automation rules.""" """Offer calendar automation rules."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Coroutine
import datetime import datetime
import logging import logging
from typing import Any from typing import Any
@ -47,7 +48,7 @@ class CalendarEventListener:
def __init__( def __init__(
self, self,
hass: HomeAssistant, hass: HomeAssistant,
job: HassJob, job: HassJob[..., Coroutine[Any, Any, None]],
trigger_data: dict[str, Any], trigger_data: dict[str, Any],
entity: CalendarEntity, entity: CalendarEntity,
event_type: str, event_type: str,

View File

@ -66,7 +66,7 @@ class DeviceTriggerAccessory(HomeAccessory):
async def async_trigger( async def async_trigger(
self, self,
run_variables: dict, run_variables: dict[str, Any],
context: Context | None = None, context: Context | None = None,
skip_condition: bool = False, skip_condition: bool = False,
) -> None: ) -> None:

View File

@ -2,7 +2,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Callable, Mapping from collections.abc import Callable, Coroutine, Mapping
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any from typing import Any
@ -84,7 +84,9 @@ class PluggableAction:
def __init__(self, update: Callable[[], None]) -> None: def __init__(self, update: Callable[[], None]) -> None:
"""Initialize.""" """Initialize."""
self._update = update self._update = update
self._actions: dict[Any, tuple[HassJob, dict[str, Any]]] = {} self._actions: dict[
Any, tuple[HassJob[..., Coroutine[Any, Any, None]], dict[str, Any]]
] = {}
def __bool__(self): def __bool__(self):
"""Return if we have something attached.""" """Return if we have something attached."""

View File

@ -1,7 +1,7 @@
"""Support for LG webOS Smart TV.""" """Support for LG webOS Smart TV."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable, Coroutine
from contextlib import suppress from contextlib import suppress
import logging import logging
from typing import Any from typing import Any
@ -170,7 +170,10 @@ class PluggableAction:
def __init__(self) -> None: def __init__(self) -> None:
"""Initialize.""" """Initialize."""
self._actions: dict[Callable[[], None], tuple[HassJob, dict[str, Any]]] = {} self._actions: dict[
Callable[[], None],
tuple[HassJob[..., Coroutine[Any, Any, None]], dict[str, Any]],
] = {}
def __bool__(self) -> bool: def __bool__(self) -> bool:
"""Return if we have something attached.""" """Return if we have something attached."""