From 1626c53c13866cc1fb0b19db2175dcde6af5f097 Mon Sep 17 00:00:00 2001
From: Marc Mueller <30130371+cdce8p@users.noreply.github.com>
Date: Wed, 20 Jul 2022 04:11:46 +0200
Subject: [PATCH] Improve dispatcher helper typing (#75455)
* Improve dispatcher helper typing
* Code review
---
.strict-typing | 1 +
homeassistant/helpers/dispatcher.py | 14 +++++++++-----
mypy.ini | 3 +++
3 files changed, 13 insertions(+), 5 deletions(-)
diff --git a/.strict-typing b/.strict-typing
index c184990884f..9757a5f6c03 100644
--- a/.strict-typing
+++ b/.strict-typing
@@ -18,6 +18,7 @@ homeassistant.helpers.condition
homeassistant.helpers.debounce
homeassistant.helpers.deprecation
homeassistant.helpers.discovery
+homeassistant.helpers.dispatcher
homeassistant.helpers.entity
homeassistant.helpers.entity_values
homeassistant.helpers.event
diff --git a/homeassistant/helpers/dispatcher.py b/homeassistant/helpers/dispatcher.py
index 07a3e3b3b28..c7ad4fb1adf 100644
--- a/homeassistant/helpers/dispatcher.py
+++ b/homeassistant/helpers/dispatcher.py
@@ -1,7 +1,7 @@
"""Helpers for Home Assistant dispatcher & internal component/platform."""
from __future__ import annotations
-from collections.abc import Callable
+from collections.abc import Callable, Coroutine
import logging
from typing import Any
@@ -62,7 +62,9 @@ def dispatcher_send(hass: HomeAssistant, signal: str, *args: Any) -> None:
hass.loop.call_soon_threadsafe(async_dispatcher_send, hass, signal, *args)
-def _generate_job(signal: str, target: Callable[..., Any]) -> HassJob:
+def _generate_job(
+ signal: str, target: Callable[..., Any]
+) -> HassJob[..., None | Coroutine[Any, Any, None]]:
"""Generate a HassJob for a signal and target."""
return HassJob(
catch_log_exception(
@@ -84,16 +86,18 @@ def async_dispatcher_send(hass: HomeAssistant, signal: str, *args: Any) -> None:
This method must be run in the event loop.
"""
- target_list = hass.data.get(DATA_DISPATCHER, {}).get(signal, {})
+ target_list: dict[
+ Callable[..., Any], HassJob[..., None | Coroutine[Any, Any, None]] | None
+ ] = hass.data.get(DATA_DISPATCHER, {}).get(signal, {})
- run: list[HassJob] = []
+ run: list[HassJob[..., None | Coroutine[Any, Any, None]]] = []
for target, job in target_list.items():
if job is None:
job = _generate_job(signal, target)
target_list[target] = job
# Run the jobs all at the end
- # to ensure no jobs add more disptachers
+ # to ensure no jobs add more dispatchers
# which can result in the target_list
# changing size during iteration
run.append(job)
diff --git a/mypy.ini b/mypy.ini
index 6321ee04b98..d7f3db26d55 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -66,6 +66,9 @@ disallow_any_generics = true
[mypy-homeassistant.helpers.discovery]
disallow_any_generics = true
+[mypy-homeassistant.helpers.dispatcher]
+disallow_any_generics = true
+
[mypy-homeassistant.helpers.entity]
disallow_any_generics = true
--
GitLab