Use PEP 695 for function annotations with scoping (#117787)

This commit is contained in:
Marc Mueller 2024-05-20 12:43:39 +02:00 committed by GitHub
parent f50973c76c
commit 7998f874c0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 14 additions and 23 deletions

View file

@ -4,7 +4,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from datetime import timedelta from datetime import timedelta
from typing import TYPE_CHECKING, Any, TypeVar from typing import TYPE_CHECKING, Any
from pyfronius import BadStatusError, FroniusError from pyfronius import BadStatusError, FroniusError
@ -32,8 +32,6 @@ if TYPE_CHECKING:
from . import FroniusSolarNet from . import FroniusSolarNet
from .sensor import _FroniusSensorEntity from .sensor import _FroniusSensorEntity
_FroniusEntityT = TypeVar("_FroniusEntityT", bound=_FroniusSensorEntity)
class FroniusCoordinatorBase( class FroniusCoordinatorBase(
ABC, DataUpdateCoordinator[dict[SolarNetId, dict[str, Any]]] ABC, DataUpdateCoordinator[dict[SolarNetId, dict[str, Any]]]
@ -84,7 +82,7 @@ class FroniusCoordinatorBase(
return data return data
@callback @callback
def add_entities_for_seen_keys( def add_entities_for_seen_keys[_FroniusEntityT: _FroniusSensorEntity](
self, self,
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
entity_constructor: type[_FroniusEntityT], entity_constructor: type[_FroniusEntityT],

View file

@ -6,16 +6,13 @@ from collections.abc import Awaitable, Callable, Coroutine
from functools import wraps from functools import wraps
from http import HTTPStatus from http import HTTPStatus
import logging import logging
from typing import Any, Concatenate, ParamSpec, TypeVar from typing import Any, Concatenate
from aiohttp import web from aiohttp import web
import voluptuous as vol import voluptuous as vol
from .view import HomeAssistantView from .view import HomeAssistantView
_HassViewT = TypeVar("_HassViewT", bound=HomeAssistantView)
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -36,7 +33,7 @@ class RequestDataValidator:
self._schema = schema self._schema = schema
self._allow_empty = allow_empty self._allow_empty = allow_empty
def __call__( def __call__[_HassViewT: HomeAssistantView, **_P](
self, self,
method: Callable[ method: Callable[
Concatenate[_HassViewT, web.Request, dict[str, Any], _P], Concatenate[_HassViewT, web.Request, dict[str, Any], _P],

View file

@ -7,7 +7,7 @@ from datetime import timedelta
import functools import functools
import logging import logging
import threading import threading
from typing import Any, ParamSpec from typing import Any
from pilight import pilight from pilight import pilight
import voluptuous as vol 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.helpers.typing import ConfigType
from homeassistant.util import dt as dt_util from homeassistant.util import dt as dt_util
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
CONF_SEND_DELAY = "send_delay" CONF_SEND_DELAY = "send_delay"
@ -147,7 +145,7 @@ class CallRateDelayThrottle:
self._next_ts = dt_util.utcnow() self._next_ts = dt_util.utcnow()
self._schedule = functools.partial(track_point_in_utc_time, hass) 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.""" """Decorate to delay calls on a certain method."""
@functools.wraps(method) @functools.wraps(method)

View file

@ -6,7 +6,7 @@ from collections.abc import Callable
from contextlib import suppress from contextlib import suppress
import logging import logging
import string import string
from typing import Any, TypeVar, cast from typing import Any, cast
from aiohttp import web from aiohttp import web
import prometheus_client import prometheus_client
@ -61,7 +61,6 @@ from homeassistant.helpers.typing import ConfigType
from homeassistant.util.dt import as_timestamp from homeassistant.util.dt import as_timestamp
from homeassistant.util.unit_conversion import TemperatureConverter from homeassistant.util.unit_conversion import TemperatureConverter
_MetricBaseT = TypeVar("_MetricBaseT", bound=MetricWrapperBase)
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
API_ENDPOINT = "/api/prometheus" API_ENDPOINT = "/api/prometheus"
@ -286,7 +285,7 @@ class PrometheusMetrics:
except (ValueError, TypeError): except (ValueError, TypeError):
pass pass
def _metric( def _metric[_MetricBaseT: MetricWrapperBase](
self, self,
metric: str, metric: str,
factory: type[_MetricBaseT], factory: type[_MetricBaseT],

View file

@ -129,7 +129,6 @@ FINAL_WRITE_STAGE_SHUTDOWN_TIMEOUT = 60
CLOSE_STAGE_SHUTDOWN_TIMEOUT = 30 CLOSE_STAGE_SHUTDOWN_TIMEOUT = 30
_R = TypeVar("_R")
# Internal; not helpers.typing.UNDEFINED due to circular dependency # Internal; not helpers.typing.UNDEFINED due to circular dependency
_UNDEF: dict[Any, Any] = {} _UNDEF: dict[Any, Any] = {}
_SENTINEL = object() _SENTINEL = object()
@ -693,7 +692,7 @@ class HomeAssistant:
@overload @overload
@callback @callback
def _async_add_hass_job( def _async_add_hass_job[_R](
self, self,
hassjob: HassJob[..., Coroutine[Any, Any, _R]], hassjob: HassJob[..., Coroutine[Any, Any, _R]],
*args: Any, *args: Any,
@ -702,7 +701,7 @@ class HomeAssistant:
@overload @overload
@callback @callback
def _async_add_hass_job( def _async_add_hass_job[_R](
self, self,
hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R], hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R],
*args: Any, *args: Any,
@ -710,7 +709,7 @@ class HomeAssistant:
) -> asyncio.Future[_R] | None: ... ) -> asyncio.Future[_R] | None: ...
@callback @callback
def _async_add_hass_job( def _async_add_hass_job[_R](
self, self,
hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R], hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R],
*args: Any, *args: Any,
@ -882,7 +881,7 @@ class HomeAssistant:
@overload @overload
@callback @callback
def async_run_hass_job( def async_run_hass_job[_R](
self, self,
hassjob: HassJob[..., Coroutine[Any, Any, _R]], hassjob: HassJob[..., Coroutine[Any, Any, _R]],
*args: Any, *args: Any,
@ -891,7 +890,7 @@ class HomeAssistant:
@overload @overload
@callback @callback
def async_run_hass_job( def async_run_hass_job[_R](
self, self,
hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R], hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R],
*args: Any, *args: Any,
@ -899,7 +898,7 @@ class HomeAssistant:
) -> asyncio.Future[_R] | None: ... ) -> asyncio.Future[_R] | None: ...
@callback @callback
def async_run_hass_job( def async_run_hass_job[_R](
self, self,
hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R], hassjob: HassJob[..., Coroutine[Any, Any, _R] | _R],
*args: Any, *args: Any,