Various type hint related improvements (#54971)
* Avoid some implicit generic Anys * Fix hassio discovery view type hints * Fix http view result type in assert message
This commit is contained in:
parent
de354f96fe
commit
59809503d1
7 changed files with 22 additions and 16 deletions
|
@ -4,7 +4,9 @@ from __future__ import annotations
|
||||||
import re
|
import re
|
||||||
from typing import Final
|
from typing import Final
|
||||||
|
|
||||||
LEASES_REGEX: Final[re.Pattern] = re.compile(
|
# mypy: disallow-any-generics
|
||||||
|
|
||||||
|
LEASES_REGEX: Final[re.Pattern[str]] = re.compile(
|
||||||
r"(?P<ip>([0-9]{1,3}[\.]){3}[0-9]{1,3})"
|
r"(?P<ip>([0-9]{1,3}[\.]){3}[0-9]{1,3})"
|
||||||
+ r"\smac:\s(?P<mac>([0-9a-f]{2}[:-]){5}([0-9a-f]{2}))"
|
+ r"\smac:\s(?P<mac>([0-9a-f]{2}[:-]){5}([0-9a-f]{2}))"
|
||||||
+ r"\svalid\sfor:\s(?P<timevalid>(-?\d+))"
|
+ r"\svalid\sfor:\s(?P<timevalid>(-?\d+))"
|
||||||
|
|
|
@ -8,7 +8,7 @@ from aiohttp.web_exceptions import HTTPServiceUnavailable
|
||||||
from homeassistant import config_entries
|
from homeassistant import config_entries
|
||||||
from homeassistant.components.http import HomeAssistantView
|
from homeassistant.components.http import HomeAssistantView
|
||||||
from homeassistant.const import ATTR_NAME, ATTR_SERVICE, EVENT_HOMEASSISTANT_START
|
from homeassistant.const import ATTR_NAME, ATTR_SERVICE, EVENT_HOMEASSISTANT_START
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import HomeAssistant, callback
|
||||||
|
|
||||||
from .const import ATTR_ADDON, ATTR_CONFIG, ATTR_DISCOVERY, ATTR_UUID
|
from .const import ATTR_ADDON, ATTR_CONFIG, ATTR_DISCOVERY, ATTR_UUID
|
||||||
from .handler import HassioAPIError
|
from .handler import HassioAPIError
|
||||||
|
@ -17,7 +17,7 @@ _LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_setup_discovery_view(hass: HomeAssistantView, hassio):
|
def async_setup_discovery_view(hass: HomeAssistant, hassio):
|
||||||
"""Discovery setup."""
|
"""Discovery setup."""
|
||||||
hassio_discovery = HassIODiscovery(hass, hassio)
|
hassio_discovery = HassIODiscovery(hass, hassio)
|
||||||
hass.http.register_view(hassio_discovery)
|
hass.http.register_view(hassio_discovery)
|
||||||
|
@ -49,7 +49,7 @@ class HassIODiscovery(HomeAssistantView):
|
||||||
name = "api:hassio_push:discovery"
|
name = "api:hassio_push:discovery"
|
||||||
url = "/api/hassio_push/discovery/{uuid}"
|
url = "/api/hassio_push/discovery/{uuid}"
|
||||||
|
|
||||||
def __init__(self, hass: HomeAssistantView, hassio):
|
def __init__(self, hass: HomeAssistant, hassio):
|
||||||
"""Initialize WebView."""
|
"""Initialize WebView."""
|
||||||
self.hass = hass
|
self.hass = hass
|
||||||
self.hassio = hassio
|
self.hassio = hassio
|
||||||
|
|
|
@ -158,7 +158,7 @@ def request_handler_factory(
|
||||||
else:
|
else:
|
||||||
assert (
|
assert (
|
||||||
False
|
False
|
||||||
), f"Result should be None, string, bytes or Response. Got: {result}"
|
), f"Result should be None, string, bytes or StreamResponse. Got: {result}"
|
||||||
|
|
||||||
return web.Response(body=bresult, status=status_code)
|
return web.Response(body=bresult, status=status_code)
|
||||||
|
|
||||||
|
|
|
@ -69,6 +69,8 @@ from .trace import (
|
||||||
trace_stack_top,
|
trace_stack_top,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# mypy: disallow-any-generics
|
||||||
|
|
||||||
FROM_CONFIG_FORMAT = "{}_from_config"
|
FROM_CONFIG_FORMAT = "{}_from_config"
|
||||||
ASYNC_FROM_CONFIG_FORMAT = "async_{}_from_config"
|
ASYNC_FROM_CONFIG_FORMAT = "async_{}_from_config"
|
||||||
|
|
||||||
|
@ -113,7 +115,7 @@ def condition_trace_update_result(**kwargs: Any) -> None:
|
||||||
|
|
||||||
|
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def trace_condition(variables: TemplateVarsType) -> Generator:
|
def trace_condition(variables: TemplateVarsType) -> Generator[TraceElement, None, None]:
|
||||||
"""Trace condition evaluation."""
|
"""Trace condition evaluation."""
|
||||||
should_pop = True
|
should_pop = True
|
||||||
trace_element = trace_stack_top(trace_stack_cv)
|
trace_element = trace_stack_top(trace_stack_cv)
|
||||||
|
|
|
@ -71,8 +71,8 @@ class TrackStates:
|
||||||
"""
|
"""
|
||||||
|
|
||||||
all_states: bool
|
all_states: bool
|
||||||
entities: set
|
entities: set[str]
|
||||||
domains: set
|
domains: set[str]
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
@ -394,7 +394,7 @@ def async_track_entity_registry_updated_event(
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def _async_dispatch_domain_event(
|
def _async_dispatch_domain_event(
|
||||||
hass: HomeAssistant, event: Event, callbacks: dict[str, list]
|
hass: HomeAssistant, event: Event, callbacks: dict[str, list[HassJob]]
|
||||||
) -> None:
|
) -> None:
|
||||||
domain = split_entity_id(event.data["entity_id"])[0]
|
domain = split_entity_id(event.data["entity_id"])[0]
|
||||||
|
|
||||||
|
@ -620,7 +620,7 @@ class _TrackStateChangeFiltered:
|
||||||
self._listeners.pop(listener_name)()
|
self._listeners.pop(listener_name)()
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def _setup_entities_listener(self, domains: set, entities: set) -> None:
|
def _setup_entities_listener(self, domains: set[str], entities: set[str]) -> None:
|
||||||
if domains:
|
if domains:
|
||||||
entities = entities.copy()
|
entities = entities.copy()
|
||||||
entities.update(self.hass.states.async_entity_ids(domains))
|
entities.update(self.hass.states.async_entity_ids(domains))
|
||||||
|
@ -634,7 +634,7 @@ class _TrackStateChangeFiltered:
|
||||||
)
|
)
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def _setup_domains_listener(self, domains: set) -> None:
|
def _setup_domains_listener(self, domains: set[str]) -> None:
|
||||||
if not domains:
|
if not domains:
|
||||||
return
|
return
|
||||||
|
|
||||||
|
|
|
@ -21,7 +21,7 @@ class TraceElement:
|
||||||
self._child_run_id: str | None = None
|
self._child_run_id: str | None = None
|
||||||
self._error: Exception | None = None
|
self._error: Exception | None = None
|
||||||
self.path: str = path
|
self.path: str = path
|
||||||
self._result: dict | None = None
|
self._result: dict[str, Any] | None = None
|
||||||
self.reuse_by_child = False
|
self.reuse_by_child = False
|
||||||
self._timestamp = dt_util.utcnow()
|
self._timestamp = dt_util.utcnow()
|
||||||
|
|
||||||
|
|
|
@ -17,6 +17,8 @@ from homeassistant.loader import (
|
||||||
from homeassistant.util.async_ import gather_with_concurrency
|
from homeassistant.util.async_ import gather_with_concurrency
|
||||||
from homeassistant.util.json import load_json
|
from homeassistant.util.json import load_json
|
||||||
|
|
||||||
|
# mypy: disallow-any-generics
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
|
||||||
TRANSLATION_LOAD_LOCK = "translation_load_lock"
|
TRANSLATION_LOAD_LOCK = "translation_load_lock"
|
||||||
|
@ -24,7 +26,7 @@ TRANSLATION_FLATTEN_CACHE = "translation_flatten_cache"
|
||||||
LOCALE_EN = "en"
|
LOCALE_EN = "en"
|
||||||
|
|
||||||
|
|
||||||
def recursive_flatten(prefix: Any, data: dict) -> dict[str, Any]:
|
def recursive_flatten(prefix: Any, data: dict[str, Any]) -> dict[str, Any]:
|
||||||
"""Return a flattened representation of dict data."""
|
"""Return a flattened representation of dict data."""
|
||||||
output = {}
|
output = {}
|
||||||
for key, value in data.items():
|
for key, value in data.items():
|
||||||
|
@ -212,7 +214,7 @@ class _TranslationCache:
|
||||||
self,
|
self,
|
||||||
language: str,
|
language: str,
|
||||||
category: str,
|
category: str,
|
||||||
components: set,
|
components: set[str],
|
||||||
) -> list[dict[str, dict[str, Any]]]:
|
) -> list[dict[str, dict[str, Any]]]:
|
||||||
"""Load resources into the cache."""
|
"""Load resources into the cache."""
|
||||||
components_to_load = components - self.loaded.setdefault(language, set())
|
components_to_load = components - self.loaded.setdefault(language, set())
|
||||||
|
@ -224,7 +226,7 @@ class _TranslationCache:
|
||||||
|
|
||||||
return [cached.get(component, {}).get(category, {}) for component in components]
|
return [cached.get(component, {}).get(category, {}) for component in components]
|
||||||
|
|
||||||
async def _async_load(self, language: str, components: set) -> None:
|
async def _async_load(self, language: str, components: set[str]) -> None:
|
||||||
"""Populate the cache for a given set of components."""
|
"""Populate the cache for a given set of components."""
|
||||||
_LOGGER.debug(
|
_LOGGER.debug(
|
||||||
"Cache miss for %s: %s",
|
"Cache miss for %s: %s",
|
||||||
|
@ -247,7 +249,7 @@ class _TranslationCache:
|
||||||
def _build_category_cache(
|
def _build_category_cache(
|
||||||
self,
|
self,
|
||||||
language: str,
|
language: str,
|
||||||
components: set,
|
components: set[str],
|
||||||
translation_strings: dict[str, dict[str, Any]],
|
translation_strings: dict[str, dict[str, Any]],
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Extract resources into the cache."""
|
"""Extract resources into the cache."""
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue