From 7998f874c09afc0d0537279d92aefb92da6fc573 Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Mon, 20 May 2024 12:43:39 +0200 Subject: [PATCH] Use PEP 695 for function annotations with scoping (#117787) --- homeassistant/components/fronius/coordinator.py | 6 ++---- homeassistant/components/http/data_validator.py | 7 ++----- homeassistant/components/pilight/__init__.py | 6 ++---- homeassistant/components/prometheus/__init__.py | 5 ++--- homeassistant/core.py | 13 ++++++------- 5 files changed, 14 insertions(+), 23 deletions(-) diff --git a/homeassistant/components/fronius/coordinator.py b/homeassistant/components/fronius/coordinator.py index 71ecb4e762e..c3dea123a77 100644 --- a/homeassistant/components/fronius/coordinator.py +++ b/homeassistant/components/fronius/coordinator.py @@ -4,7 +4,7 @@ from __future__ import annotations from abc import ABC, abstractmethod from datetime import timedelta -from typing import TYPE_CHECKING, Any, TypeVar +from typing import TYPE_CHECKING, Any from pyfronius import BadStatusError, FroniusError @@ -32,8 +32,6 @@ if TYPE_CHECKING: from . import FroniusSolarNet from .sensor import _FroniusSensorEntity - _FroniusEntityT = TypeVar("_FroniusEntityT", bound=_FroniusSensorEntity) - class FroniusCoordinatorBase( ABC, DataUpdateCoordinator[dict[SolarNetId, dict[str, Any]]] @@ -84,7 +82,7 @@ class FroniusCoordinatorBase( return data @callback - def add_entities_for_seen_keys( + def add_entities_for_seen_keys[_FroniusEntityT: _FroniusSensorEntity]( self, async_add_entities: AddEntitiesCallback, entity_constructor: type[_FroniusEntityT], diff --git a/homeassistant/components/http/data_validator.py b/homeassistant/components/http/data_validator.py index e1ba1caae56..b2f6496a77b 100644 --- a/homeassistant/components/http/data_validator.py +++ b/homeassistant/components/http/data_validator.py @@ -6,16 +6,13 @@ from collections.abc import Awaitable, Callable, Coroutine from functools import wraps from http import HTTPStatus import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from aiohttp import web import voluptuous as vol from .view import HomeAssistantView -_HassViewT = TypeVar("_HassViewT", bound=HomeAssistantView) -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) @@ -36,7 +33,7 @@ class RequestDataValidator: self._schema = schema self._allow_empty = allow_empty - def __call__( + def __call__[_HassViewT: HomeAssistantView, **_P]( self, method: Callable[ Concatenate[_HassViewT, web.Request, dict[str, Any], _P], diff --git a/homeassistant/components/pilight/__init__.py b/homeassistant/components/pilight/__init__.py index 1f1eee0c92a..21d5603e4c2 100644 --- a/homeassistant/components/pilight/__init__.py +++ b/homeassistant/components/pilight/__init__.py @@ -7,7 +7,7 @@ from datetime import timedelta import functools import logging import threading -from typing import Any, ParamSpec +from typing import Any from pilight import pilight import voluptuous as vol @@ -26,8 +26,6 @@ from homeassistant.helpers.event import track_point_in_utc_time from homeassistant.helpers.typing import ConfigType from homeassistant.util import dt as dt_util -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) CONF_SEND_DELAY = "send_delay" @@ -147,7 +145,7 @@ class CallRateDelayThrottle: self._next_ts = dt_util.utcnow() self._schedule = functools.partial(track_point_in_utc_time, hass) - def limited(self, method: Callable[_P, Any]) -> Callable[_P, None]: + def limited[**_P](self, method: Callable[_P, Any]) -> Callable[_P, None]: """Decorate to delay calls on a certain method.""" @functools.wraps(method) diff --git a/homeassistant/components/prometheus/__init__.py b/homeassistant/components/prometheus/__init__.py index c02cbeabd84..2159656f129 100644 --- a/homeassistant/components/prometheus/__init__.py +++ b/homeassistant/components/prometheus/__init__.py @@ -6,7 +6,7 @@ from collections.abc import Callable from contextlib import suppress import logging import string -from typing import Any, TypeVar, cast +from typing import Any, cast from aiohttp import web import prometheus_client @@ -61,7 +61,6 @@ from homeassistant.helpers.typing import ConfigType from homeassistant.util.dt import as_timestamp from homeassistant.util.unit_conversion import TemperatureConverter -_MetricBaseT = TypeVar("_MetricBaseT", bound=MetricWrapperBase) _LOGGER = logging.getLogger(__name__) API_ENDPOINT = "/api/prometheus" @@ -286,7 +285,7 @@ class PrometheusMetrics: except (ValueError, TypeError): pass - def _metric( + def _metric[_MetricBaseT: MetricWrapperBase]( self, metric: str, factory: type[_MetricBaseT], diff --git a/homeassistant/core.py b/homeassistant/core.py index 23430912402..ca82b46bb87 100644 --- a/homeassistant/core.py +++ b/homeassistant/core.py @@ -129,7 +129,6 @@ FINAL_WRITE_STAGE_SHUTDOWN_TIMEOUT = 60 CLOSE_STAGE_SHUTDOWN_TIMEOUT = 30 -_R = TypeVar("_R") # Internal; not helpers.typing.UNDEFINED due to circular dependency _UNDEF: dict[Any, Any] = {} _SENTINEL = object() @@ -693,7 +692,7 @@ class HomeAssistant: @overload @callback - def _async_add_hass_job( + def _async_add_hass_job[_R]( self, hassjob: HassJob[..., Coroutine[Any, Any, _R]], *args: Any, @@ -702,7 +701,7 @@ class HomeAssistant: @overload @callback - def _async_add_hass_job( + def _async_add_hass_job[_R]( self, hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R], *args: Any, @@ -710,7 +709,7 @@ class HomeAssistant: ) -> asyncio.Future[_R] | None: ... @callback - def _async_add_hass_job( + def _async_add_hass_job[_R]( self, hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R], *args: Any, @@ -882,7 +881,7 @@ class HomeAssistant: @overload @callback - def async_run_hass_job( + def async_run_hass_job[_R]( self, hassjob: HassJob[..., Coroutine[Any, Any, _R]], *args: Any, @@ -891,7 +890,7 @@ class HomeAssistant: @overload @callback - def async_run_hass_job( + def async_run_hass_job[_R]( self, hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R], *args: Any, @@ -899,7 +898,7 @@ class HomeAssistant: ) -> asyncio.Future[_R] | None: ... @callback - def async_run_hass_job( + def async_run_hass_job[_R]( self, hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R], *args: Any,