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

Use a normal tuple for the EventBus jobs (#95731)

Browse Source
This commit is contained in:
J. Nick Koston 2023-07-02 16:35:57 -05:00 committed by GitHub
parent c1b8e4a3e5
commit 0ff3836083
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 12 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
homeassistant/core.py
View File

@ -31,7 +31,6 @@ from typing import (
TYPE_CHECKING, TYPE_CHECKING,
Any, Any,
Generic, Generic,
NamedTuple,
ParamSpec, ParamSpec,
TypeVar, TypeVar,
cast, cast,
@ -964,12 +963,11 @@ class Event:
return f"<Event {self.event_type}[{str(self.origin)[0]}]>" return f"<Event {self.event_type}[{str(self.origin)[0]}]>"
class _FilterableJob(NamedTuple): _FilterableJobType = tuple[
"""Event listener job to be executed with optional filter.""" HassJob[[Event], Coroutine[Any, Any, None] | None], # job
Callable[[Event], bool] | None, # event_filter
job: HassJob[[Event], Coroutine[Any, Any, None] | None] bool, # run_immediately
event_filter: Callable[[Event], bool] | None ]
run_immediately: bool
class EventBus: class EventBus:
@ -977,8 +975,8 @@ class EventBus:
def __init__(self, hass: HomeAssistant) -> None: def __init__(self, hass: HomeAssistant) -> None:
"""Initialize a new event bus.""" """Initialize a new event bus."""
self._listeners: dict[str, list[_FilterableJob]] = {} self._listeners: dict[str, list[_FilterableJobType]] = {}
self._match_all_listeners: list[_FilterableJob] = [] self._match_all_listeners: list[_FilterableJobType] = []
self._listeners[MATCH_ALL] = self._match_all_listeners self._listeners[MATCH_ALL] = self._match_all_listeners
self._hass = hass self._hass = hass
@ -1105,14 +1103,12 @@ class EventBus:
raise HomeAssistantError(f"Event listener {listener} is not a callback") raise HomeAssistantError(f"Event listener {listener} is not a callback")
return self._async_listen_filterable_job( return self._async_listen_filterable_job(
event_type, event_type,
_FilterableJob( (HassJob(listener, f"listen {event_type}"), event_filter, run_immediately),
HassJob(listener, f"listen {event_type}"), event_filter, run_immediately
),
) )
@callback @callback
def _async_listen_filterable_job( def _async_listen_filterable_job(
self, event_type: str, filterable_job: _FilterableJob self, event_type: str, filterable_job: _FilterableJobType
) -> CALLBACK_TYPE: ) -> CALLBACK_TYPE:
self._listeners.setdefault(event_type, []).append(filterable_job) self._listeners.setdefault(event_type, []).append(filterable_job)
@ -1159,7 +1155,7 @@ class EventBus:
This method must be run in the event loop. This method must be run in the event loop.
""" """
filterable_job: _FilterableJob | None = None filterable_job: _FilterableJobType | None = None
@callback @callback
def _onetime_listener(event: Event) -> None: def _onetime_listener(event: Event) -> None:
@ -1181,7 +1177,7 @@ class EventBus:
_onetime_listener, listener, ("__name__", "__qualname__", "__module__"), [] _onetime_listener, listener, ("__name__", "__qualname__", "__module__"), []
) )
filterable_job = _FilterableJob( filterable_job = (
HassJob(_onetime_listener, f"onetime listen {event_type} {listener}"), HassJob(_onetime_listener, f"onetime listen {event_type} {listener}"),
None, None,
False, False,
@ -1191,7 +1187,7 @@ class EventBus:
@callback @callback
def _async_remove_listener( def _async_remove_listener(
self, event_type: str, filterable_job: _FilterableJob self, event_type: str, filterable_job: _FilterableJobType
) -> None: ) -> None:
"""Remove a listener of a specific event_type. """Remove a listener of a specific event_type.