mirror of
https://github.com/home-assistant/core.git
synced 2025-07-18 10:47:10 +00:00
Ensure hass is typed (#87068)
* Ensure hass is typed * Adjust pilight * Adjust homeassistant scene * Adjust hassio * Adjust gree * Adjust google_maps * Adjust energyzero * Adjust harmony * Adjust mobile_app
This commit is contained in:
parent
3d557b5583
commit
f6c76372ce
@ -13,6 +13,7 @@ from energyzero import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
from homeassistant.config_entries import ConfigEntry
|
from homeassistant.config_entries import ConfigEntry
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||||
from homeassistant.util import dt
|
from homeassistant.util import dt
|
||||||
@ -33,7 +34,7 @@ class EnergyZeroDataUpdateCoordinator(DataUpdateCoordinator[EnergyZeroData]):
|
|||||||
|
|
||||||
config_entry: ConfigEntry
|
config_entry: ConfigEntry
|
||||||
|
|
||||||
def __init__(self, hass) -> None:
|
def __init__(self, hass: HomeAssistant) -> None:
|
||||||
"""Initialize global EnergyZero data updater."""
|
"""Initialize global EnergyZero data updater."""
|
||||||
super().__init__(
|
super().__init__(
|
||||||
hass,
|
hass,
|
||||||
|
@ -61,7 +61,9 @@ def setup_scanner(
|
|||||||
class GoogleMapsScanner:
|
class GoogleMapsScanner:
|
||||||
"""Representation of an Google Maps location sharing account."""
|
"""Representation of an Google Maps location sharing account."""
|
||||||
|
|
||||||
def __init__(self, hass, config: ConfigType, see: SeeCallback) -> None:
|
def __init__(
|
||||||
|
self, hass: HomeAssistant, config: ConfigType, see: SeeCallback
|
||||||
|
) -> None:
|
||||||
"""Initialize the scanner."""
|
"""Initialize the scanner."""
|
||||||
self.see = see
|
self.see = see
|
||||||
self.username = config[CONF_USERNAME]
|
self.username = config[CONF_USERNAME]
|
||||||
|
@ -71,7 +71,7 @@ class DeviceDataUpdateCoordinator(DataUpdateCoordinator):
|
|||||||
class DiscoveryService(Listener):
|
class DiscoveryService(Listener):
|
||||||
"""Discovery event handler for gree devices."""
|
"""Discovery event handler for gree devices."""
|
||||||
|
|
||||||
def __init__(self, hass) -> None:
|
def __init__(self, hass: HomeAssistant) -> None:
|
||||||
"""Initialize discovery service."""
|
"""Initialize discovery service."""
|
||||||
super().__init__()
|
super().__init__()
|
||||||
self.hass = hass
|
self.hass = hass
|
||||||
|
@ -9,6 +9,7 @@ from aioharmony.const import ClientCallbackType, SendCommandDevice
|
|||||||
import aioharmony.exceptions as aioexc
|
import aioharmony.exceptions as aioexc
|
||||||
from aioharmony.harmonyapi import HarmonyAPI as HarmonyClient
|
from aioharmony.harmonyapi import HarmonyAPI as HarmonyClient
|
||||||
|
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
from homeassistant.exceptions import ConfigEntryNotReady
|
from homeassistant.exceptions import ConfigEntryNotReady
|
||||||
from homeassistant.helpers.entity import DeviceInfo
|
from homeassistant.helpers.entity import DeviceInfo
|
||||||
|
|
||||||
@ -23,7 +24,9 @@ class HarmonyData(HarmonySubscriberMixin):
|
|||||||
|
|
||||||
_client: HarmonyClient
|
_client: HarmonyClient
|
||||||
|
|
||||||
def __init__(self, hass, address: str, name: str, unique_id: str | None) -> None:
|
def __init__(
|
||||||
|
self, hass: HomeAssistant, address: str, name: str, unique_id: str | None
|
||||||
|
) -> None:
|
||||||
"""Initialize a data object."""
|
"""Initialize a data object."""
|
||||||
super().__init__(hass)
|
super().__init__(hass)
|
||||||
self._name = name
|
self._name = name
|
||||||
|
@ -23,6 +23,7 @@ from multidict import istr
|
|||||||
|
|
||||||
from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
|
from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
|
||||||
from homeassistant.components.onboarding import async_is_onboarded
|
from homeassistant.components.onboarding import async_is_onboarded
|
||||||
|
from homeassistant.core import HomeAssistant
|
||||||
|
|
||||||
from .const import X_HASS_IS_ADMIN, X_HASS_USER_ID
|
from .const import X_HASS_IS_ADMIN, X_HASS_USER_ID
|
||||||
|
|
||||||
@ -165,7 +166,7 @@ def _get_timeout(path: str) -> ClientTimeout:
|
|||||||
return ClientTimeout(connect=10, total=300)
|
return ClientTimeout(connect=10, total=300)
|
||||||
|
|
||||||
|
|
||||||
def _need_auth(hass, path: str) -> bool:
|
def _need_auth(hass: HomeAssistant, path: str) -> bool:
|
||||||
"""Return if a path need authentication."""
|
"""Return if a path need authentication."""
|
||||||
if not async_is_onboarded(hass) and NO_AUTH_ONBOARDING.match(path):
|
if not async_is_onboarded(hass) and NO_AUTH_ONBOARDING.match(path):
|
||||||
return False
|
return False
|
||||||
|
@ -273,7 +273,7 @@ async def async_setup_platform(
|
|||||||
|
|
||||||
|
|
||||||
def _process_scenes_config(
|
def _process_scenes_config(
|
||||||
hass, async_add_entities: AddEntitiesCallback, config: dict[str, Any]
|
hass: HomeAssistant, async_add_entities: AddEntitiesCallback, config: dict[str, Any]
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Process multiple scenes and add them."""
|
"""Process multiple scenes and add them."""
|
||||||
# Check empty list
|
# Check empty list
|
||||||
|
@ -3,7 +3,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
from typing import TYPE_CHECKING
|
from typing import TYPE_CHECKING
|
||||||
|
|
||||||
from homeassistant.core import callback
|
from homeassistant.core import HomeAssistant, callback
|
||||||
|
|
||||||
from .const import (
|
from .const import (
|
||||||
ATTR_APP_DATA,
|
ATTR_APP_DATA,
|
||||||
@ -21,7 +21,7 @@ if TYPE_CHECKING:
|
|||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def webhook_id_from_device_id(hass, device_id: str) -> str | None:
|
def webhook_id_from_device_id(hass: HomeAssistant, device_id: str) -> str | None:
|
||||||
"""Get webhook ID from device ID."""
|
"""Get webhook ID from device ID."""
|
||||||
if DOMAIN not in hass.data:
|
if DOMAIN not in hass.data:
|
||||||
return None
|
return None
|
||||||
@ -34,7 +34,7 @@ def webhook_id_from_device_id(hass, device_id: str) -> str | None:
|
|||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def supports_push(hass, webhook_id: str) -> bool:
|
def supports_push(hass: HomeAssistant, webhook_id: str) -> bool:
|
||||||
"""Return if push notifications is supported."""
|
"""Return if push notifications is supported."""
|
||||||
config_entry = hass.data[DOMAIN][DATA_CONFIG_ENTRIES][webhook_id]
|
config_entry = hass.data[DOMAIN][DATA_CONFIG_ENTRIES][webhook_id]
|
||||||
app_data = config_entry.data[ATTR_APP_DATA]
|
app_data = config_entry.data[ATTR_APP_DATA]
|
||||||
@ -44,7 +44,7 @@ def supports_push(hass, webhook_id: str) -> bool:
|
|||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def get_notify_service(hass, webhook_id: str) -> str | None:
|
def get_notify_service(hass: HomeAssistant, webhook_id: str) -> str | None:
|
||||||
"""Return the notify service for this webhook ID."""
|
"""Return the notify service for this webhook ID."""
|
||||||
notify_service: MobileAppNotificationService = hass.data[DOMAIN][DATA_NOTIFY]
|
notify_service: MobileAppNotificationService = hass.data[DOMAIN][DATA_NOTIFY]
|
||||||
|
|
||||||
|
@ -141,7 +141,7 @@ class CallRateDelayThrottle:
|
|||||||
it should not block the mainloop.
|
it should not block the mainloop.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, hass, delay_seconds: float) -> None:
|
def __init__(self, hass: HomeAssistant, delay_seconds: float) -> None:
|
||||||
"""Initialize the delay handler."""
|
"""Initialize the delay handler."""
|
||||||
self._delay = timedelta(seconds=max(0.0, delay_seconds))
|
self._delay = timedelta(seconds=max(0.0, delay_seconds))
|
||||||
self._queue: list[Callable[[Any], None]] = []
|
self._queue: list[Callable[[Any], None]] = []
|
||||||
|
@ -18,6 +18,11 @@ if TYPE_CHECKING:
|
|||||||
# pre-commit should still work on out of date environments
|
# pre-commit should still work on out of date environments
|
||||||
from astroid.typing import InferenceResult
|
from astroid.typing import InferenceResult
|
||||||
|
|
||||||
|
_COMMON_ARGUMENTS: dict[str, list[str]] = {
|
||||||
|
"hass": ["HomeAssistant", "HomeAssistant | None"]
|
||||||
|
}
|
||||||
|
_PLATFORMS: set[str] = {platform.value for platform in Platform}
|
||||||
|
|
||||||
|
|
||||||
class _Special(Enum):
|
class _Special(Enum):
|
||||||
"""Sentinel values."""
|
"""Sentinel values."""
|
||||||
@ -25,9 +30,6 @@ class _Special(Enum):
|
|||||||
UNDEFINED = 1
|
UNDEFINED = 1
|
||||||
|
|
||||||
|
|
||||||
_PLATFORMS: set[str] = {platform.value for platform in Platform}
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class TypeHintMatch:
|
class TypeHintMatch:
|
||||||
"""Class for pattern matching."""
|
"""Class for pattern matching."""
|
||||||
@ -2911,6 +2913,16 @@ class HassTypeHintChecker(BaseChecker): # type: ignore[misc]
|
|||||||
|
|
||||||
self._class_matchers.reverse()
|
self._class_matchers.reverse()
|
||||||
|
|
||||||
|
def _ignore_function(
|
||||||
|
self, node: nodes.FunctionDef, annotations: list[nodes.NodeNG | None]
|
||||||
|
) -> bool:
|
||||||
|
"""Check if we can skip the function validation."""
|
||||||
|
return (
|
||||||
|
self.linter.config.ignore_missing_annotations
|
||||||
|
and node.returns is None
|
||||||
|
and not _has_valid_annotations(annotations)
|
||||||
|
)
|
||||||
|
|
||||||
def visit_classdef(self, node: nodes.ClassDef) -> None:
|
def visit_classdef(self, node: nodes.ClassDef) -> None:
|
||||||
"""Apply relevant type hint checks on a ClassDef node."""
|
"""Apply relevant type hint checks on a ClassDef node."""
|
||||||
ancestor: nodes.ClassDef
|
ancestor: nodes.ClassDef
|
||||||
@ -2932,34 +2944,55 @@ class HassTypeHintChecker(BaseChecker): # type: ignore[misc]
|
|||||||
cached_methods: list[nodes.FunctionDef] = list(node.mymethods())
|
cached_methods: list[nodes.FunctionDef] = list(node.mymethods())
|
||||||
for match in matches:
|
for match in matches:
|
||||||
for function_node in cached_methods:
|
for function_node in cached_methods:
|
||||||
if function_node.name in checked_class_methods:
|
if (
|
||||||
|
function_node.name in checked_class_methods
|
||||||
|
or not match.need_to_check_function(function_node)
|
||||||
|
):
|
||||||
continue
|
continue
|
||||||
if match.need_to_check_function(function_node):
|
|
||||||
self._check_function(function_node, match)
|
annotations = _get_all_annotations(function_node)
|
||||||
|
if self._ignore_function(function_node, annotations):
|
||||||
|
continue
|
||||||
|
|
||||||
|
self._check_function(function_node, match, annotations)
|
||||||
checked_class_methods.add(function_node.name)
|
checked_class_methods.add(function_node.name)
|
||||||
|
|
||||||
def visit_functiondef(self, node: nodes.FunctionDef) -> None:
|
def visit_functiondef(self, node: nodes.FunctionDef) -> None:
|
||||||
"""Apply relevant type hint checks on a FunctionDef node."""
|
"""Apply relevant type hint checks on a FunctionDef node."""
|
||||||
|
annotations = _get_all_annotations(node)
|
||||||
|
if self._ignore_function(node, annotations):
|
||||||
|
return
|
||||||
|
|
||||||
|
# Check that common arguments are correctly typed.
|
||||||
|
for arg_name, expected_type in _COMMON_ARGUMENTS.items():
|
||||||
|
arg_node, annotation = _get_named_annotation(node, arg_name)
|
||||||
|
if arg_node and not _is_valid_type(expected_type, annotation):
|
||||||
|
self.add_message(
|
||||||
|
"hass-argument-type",
|
||||||
|
node=arg_node,
|
||||||
|
args=(arg_name, expected_type, node.name),
|
||||||
|
)
|
||||||
|
|
||||||
|
# Check function matchers.
|
||||||
for match in self._function_matchers:
|
for match in self._function_matchers:
|
||||||
if not match.need_to_check_function(node) or node.is_method():
|
if not match.need_to_check_function(node) or node.is_method():
|
||||||
continue
|
continue
|
||||||
self._check_function(node, match)
|
self._check_function(node, match, annotations)
|
||||||
|
|
||||||
visit_asyncfunctiondef = visit_functiondef
|
visit_asyncfunctiondef = visit_functiondef
|
||||||
|
|
||||||
def _check_function(self, node: nodes.FunctionDef, match: TypeHintMatch) -> None:
|
def _check_function(
|
||||||
# Check that at least one argument is annotated.
|
self,
|
||||||
annotations = _get_all_annotations(node)
|
node: nodes.FunctionDef,
|
||||||
if (
|
match: TypeHintMatch,
|
||||||
self.linter.config.ignore_missing_annotations
|
annotations: list[nodes.NodeNG | None],
|
||||||
and node.returns is None
|
) -> None:
|
||||||
and not _has_valid_annotations(annotations)
|
|
||||||
):
|
|
||||||
return
|
|
||||||
|
|
||||||
# Check that all positional arguments are correctly annotated.
|
# Check that all positional arguments are correctly annotated.
|
||||||
if match.arg_types:
|
if match.arg_types:
|
||||||
for key, expected_type in match.arg_types.items():
|
for key, expected_type in match.arg_types.items():
|
||||||
|
if node.args.args[key].name in _COMMON_ARGUMENTS:
|
||||||
|
# It has already been checked, avoid double-message
|
||||||
|
continue
|
||||||
if not _is_valid_type(expected_type, annotations[key]):
|
if not _is_valid_type(expected_type, annotations[key]):
|
||||||
self.add_message(
|
self.add_message(
|
||||||
"hass-argument-type",
|
"hass-argument-type",
|
||||||
@ -2970,6 +3003,9 @@ class HassTypeHintChecker(BaseChecker): # type: ignore[misc]
|
|||||||
# Check that all keyword arguments are correctly annotated.
|
# Check that all keyword arguments are correctly annotated.
|
||||||
if match.named_arg_types is not None:
|
if match.named_arg_types is not None:
|
||||||
for arg_name, expected_type in match.named_arg_types.items():
|
for arg_name, expected_type in match.named_arg_types.items():
|
||||||
|
if arg_name in _COMMON_ARGUMENTS:
|
||||||
|
# It has already been checked, avoid double-message
|
||||||
|
continue
|
||||||
arg_node, annotation = _get_named_annotation(node, arg_name)
|
arg_node, annotation = _get_named_annotation(node, arg_name)
|
||||||
if arg_node and not _is_valid_type(expected_type, annotation):
|
if arg_node and not _is_valid_type(expected_type, annotation):
|
||||||
self.add_message(
|
self.add_message(
|
||||||
|
Loading…
x
Reference in New Issue
Block a user