diff --git a/homeassistant/helpers/deprecation.py b/homeassistant/helpers/deprecation.py index 93520866142..79dd436db95 100644 --- a/homeassistant/helpers/deprecation.py +++ b/homeassistant/helpers/deprecation.py @@ -8,14 +8,10 @@ from enum import Enum import functools import inspect import logging -from typing import Any, NamedTuple, ParamSpec, TypeVar - -_ObjectT = TypeVar("_ObjectT", bound=object) -_R = TypeVar("_R") -_P = ParamSpec("_P") +from typing import Any, NamedTuple -def deprecated_substitute( +def deprecated_substitute[_ObjectT: object]( substitute_name: str, ) -> Callable[[Callable[[_ObjectT], Any]], Callable[[_ObjectT], Any]]: """Help migrate properties to new names. @@ -92,7 +88,7 @@ def get_deprecated( return config.get(new_name, default) -def deprecated_class( +def deprecated_class[**_P, _R]( replacement: str, *, breaks_in_ha_version: str | None = None ) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]: """Mark class as deprecated and provide a replacement class to be used instead. @@ -117,7 +113,7 @@ def deprecated_class( return deprecated_decorator -def deprecated_function( +def deprecated_function[**_P, _R]( replacement: str, *, breaks_in_ha_version: str | None = None ) -> Callable[[Callable[_P, _R]], Callable[_P, _R]]: """Mark function as deprecated and provide a replacement to be used instead. diff --git a/homeassistant/helpers/entity.py b/homeassistant/helpers/entity.py index 54fd1aafaeb..c6f18314012 100644 --- a/homeassistant/helpers/entity.py +++ b/homeassistant/helpers/entity.py @@ -16,16 +16,7 @@ from operator import attrgetter import sys import time from types import FunctionType -from typing import ( - TYPE_CHECKING, - Any, - Final, - Literal, - NotRequired, - TypedDict, - TypeVar, - final, -) +from typing import TYPE_CHECKING, Any, Final, Literal, NotRequired, TypedDict, final import voluptuous as vol @@ -79,8 +70,6 @@ timer = time.time if TYPE_CHECKING: from .entity_platform import EntityPlatform -_T = TypeVar("_T") - _LOGGER = logging.getLogger(__name__) SLOW_UPDATE_WARNING = 10 DATA_ENTITY_SOURCE = "entity_info" @@ -1603,7 +1592,7 @@ class Entity( return f"" return f"" - async def async_request_call(self, coro: Coroutine[Any, Any, _T]) -> _T: + async def async_request_call[_T](self, coro: Coroutine[Any, Any, _T]) -> _T: """Process request batched.""" if self.parallel_updates: await self.parallel_updates.acquire() diff --git a/homeassistant/helpers/entity_registry.py b/homeassistant/helpers/entity_registry.py index 2964c55af74..1c43c8e7ec9 100644 --- a/homeassistant/helpers/entity_registry.py +++ b/homeassistant/helpers/entity_registry.py @@ -16,7 +16,7 @@ from enum import StrEnum from functools import cached_property import logging import time -from typing import TYPE_CHECKING, Any, Literal, NotRequired, TypedDict, TypeVar +from typing import TYPE_CHECKING, Any, Literal, NotRequired, TypedDict import attr import voluptuous as vol @@ -65,8 +65,6 @@ from .typing import UNDEFINED, UndefinedType if TYPE_CHECKING: from homeassistant.config_entries import ConfigEntry -T = TypeVar("T") - DATA_REGISTRY: HassKey[EntityRegistry] = HassKey("entity_registry") EVENT_ENTITY_REGISTRY_UPDATED: EventType[EventEntityRegistryUpdatedData] = EventType( "entity_registry_updated" @@ -852,7 +850,7 @@ class EntityRegistry(BaseRegistry): ): disabled_by = RegistryEntryDisabler.INTEGRATION - def none_if_undefined(value: T | UndefinedType) -> T | None: + def none_if_undefined[_T](value: _T | UndefinedType) -> _T | None: """Return None if value is UNDEFINED, otherwise return value.""" return None if value is UNDEFINED else value diff --git a/homeassistant/helpers/frame.py b/homeassistant/helpers/frame.py index 2a6e8f87a8f..321094ba8d9 100644 --- a/homeassistant/helpers/frame.py +++ b/homeassistant/helpers/frame.py @@ -12,7 +12,7 @@ import linecache import logging import sys from types import FrameType -from typing import Any, TypeVar, cast +from typing import Any, cast from homeassistant.core import HomeAssistant, async_get_hass from homeassistant.exceptions import HomeAssistantError @@ -23,8 +23,6 @@ _LOGGER = logging.getLogger(__name__) # Keep track of integrations already reported to prevent flooding _REPORTED_INTEGRATIONS: set[str] = set() -_CallableT = TypeVar("_CallableT", bound=Callable) - @dataclass(kw_only=True) class IntegrationFrame: @@ -209,7 +207,7 @@ def _report_integration( ) -def warn_use(func: _CallableT, what: str) -> _CallableT: +def warn_use[_CallableT: Callable](func: _CallableT, what: str) -> _CallableT: """Mock a function to warn when it was about to be used.""" if asyncio.iscoroutinefunction(func): diff --git a/homeassistant/helpers/ratelimit.py b/homeassistant/helpers/ratelimit.py index 020c7c3a0d3..c9b1f21cba7 100644 --- a/homeassistant/helpers/ratelimit.py +++ b/homeassistant/helpers/ratelimit.py @@ -6,12 +6,9 @@ import asyncio from collections.abc import Callable, Hashable import logging import time -from typing import TypeVarTuple from homeassistant.core import HomeAssistant, callback -_Ts = TypeVarTuple("_Ts") - _LOGGER = logging.getLogger(__name__) @@ -52,7 +49,7 @@ class KeyedRateLimit: self._rate_limit_timers.clear() @callback - def async_schedule_action( + def async_schedule_action[*_Ts]( self, key: Hashable, rate_limit: float | None, diff --git a/homeassistant/helpers/redact.py b/homeassistant/helpers/redact.py index ad06f58a50a..6db0ab4bdd9 100644 --- a/homeassistant/helpers/redact.py +++ b/homeassistant/helpers/redact.py @@ -3,15 +3,12 @@ from __future__ import annotations from collections.abc import Callable, Iterable, Mapping -from typing import Any, TypeVar, cast, overload +from typing import Any, cast, overload from homeassistant.core import callback REDACTED = "**REDACTED**" -_T = TypeVar("_T") -_ValueT = TypeVar("_ValueT") - def partial_redact( x: str | Any, unmasked_prefix: int = 4, unmasked_suffix: int = 4 @@ -32,19 +29,19 @@ def partial_redact( @overload -def async_redact_data( # type: ignore[overload-overlap] +def async_redact_data[_ValueT]( # type: ignore[overload-overlap] data: Mapping, to_redact: Iterable[Any] | Mapping[Any, Callable[[_ValueT], _ValueT]] ) -> dict: ... @overload -def async_redact_data( +def async_redact_data[_T, _ValueT]( data: _T, to_redact: Iterable[Any] | Mapping[Any, Callable[[_ValueT], _ValueT]] ) -> _T: ... @callback -def async_redact_data( +def async_redact_data[_T, _ValueT]( data: _T, to_redact: Iterable[Any] | Mapping[Any, Callable[[_ValueT], _ValueT]] ) -> _T: """Redact sensitive data in a dict.""" diff --git a/homeassistant/helpers/script.py b/homeassistant/helpers/script.py index 7af29fb4327..c268a21758f 100644 --- a/homeassistant/helpers/script.py +++ b/homeassistant/helpers/script.py @@ -13,7 +13,7 @@ from functools import cached_property, partial import itertools import logging from types import MappingProxyType -from typing import Any, Literal, TypedDict, TypeVar, cast +from typing import Any, Literal, TypedDict, cast import async_interrupt import voluptuous as vol @@ -111,8 +111,6 @@ from .typing import UNDEFINED, ConfigType, UndefinedType # mypy: allow-untyped-calls, allow-untyped-defs, no-check-untyped-defs -_T = TypeVar("_T") - SCRIPT_MODE_PARALLEL = "parallel" SCRIPT_MODE_QUEUED = "queued" SCRIPT_MODE_RESTART = "restart" @@ -713,7 +711,9 @@ class _ScriptRun: else: wait_var["remaining"] = None - async def _async_run_long_action(self, long_task: asyncio.Task[_T]) -> _T | None: + async def _async_run_long_action[_T]( + self, long_task: asyncio.Task[_T] + ) -> _T | None: """Run a long task while monitoring for stop request.""" try: async with async_interrupt.interrupt(self._stop, ScriptStoppedError, None): diff --git a/homeassistant/helpers/service.py b/homeassistant/helpers/service.py index 1396f37e665..7d5a15f41b2 100644 --- a/homeassistant/helpers/service.py +++ b/homeassistant/helpers/service.py @@ -68,9 +68,6 @@ from .typing import ConfigType, TemplateVarsType if TYPE_CHECKING: from .entity import Entity - _EntityT = TypeVar("_EntityT", bound=Entity) - - CONF_SERVICE_ENTITY_ID = "entity_id" _LOGGER = logging.getLogger(__name__) @@ -434,7 +431,7 @@ def extract_entity_ids( @bind_hass -async def async_extract_entities( +async def async_extract_entities[_EntityT: Entity]( hass: HomeAssistant, entities: Iterable[_EntityT], service_call: ServiceCall, diff --git a/homeassistant/helpers/template.py b/homeassistant/helpers/template.py index 44b67f1c228..32c0ff244a6 100644 --- a/homeassistant/helpers/template.py +++ b/homeassistant/helpers/template.py @@ -22,17 +22,7 @@ import statistics from struct import error as StructError, pack, unpack_from import sys from types import CodeType, TracebackType -from typing import ( - Any, - Concatenate, - Literal, - NoReturn, - ParamSpec, - Self, - TypeVar, - cast, - overload, -) +from typing import Any, Concatenate, Literal, NoReturn, Self, cast, overload from urllib.parse import urlencode as urllib_urlencode import weakref @@ -134,10 +124,6 @@ _COLLECTABLE_STATE_ATTRIBUTES = { "name", } -_T = TypeVar("_T") -_R = TypeVar("_R") -_P = ParamSpec("_P") - ALL_STATES_RATE_LIMIT = 60 # seconds DOMAIN_STATES_RATE_LIMIT = 1 # seconds @@ -1217,10 +1203,10 @@ def forgiving_boolean(value: Any) -> bool | object: ... @overload -def forgiving_boolean(value: Any, default: _T) -> bool | _T: ... +def forgiving_boolean[_T](value: Any, default: _T) -> bool | _T: ... -def forgiving_boolean( +def forgiving_boolean[_T]( value: Any, default: _T | object = _SENTINEL ) -> bool | _T | object: """Try to convert value to a boolean.""" @@ -2840,7 +2826,7 @@ class TemplateEnvironment(ImmutableSandboxedEnvironment): # evaluated fresh with every execution, rather than executed # at compile time and the value stored. The context itself # can be discarded, we only need to get at the hass object. - def hassfunction( + def hassfunction[**_P, _R]( func: Callable[Concatenate[HomeAssistant, _P], _R], jinja_context: Callable[ [Callable[Concatenate[Any, _P], _R]], diff --git a/homeassistant/helpers/trace.py b/homeassistant/helpers/trace.py index 1f5aa47f4e2..17019863d9f 100644 --- a/homeassistant/helpers/trace.py +++ b/homeassistant/helpers/trace.py @@ -7,16 +7,13 @@ from collections.abc import Callable, Coroutine, Generator from contextlib import contextmanager from contextvars import ContextVar from functools import wraps -from typing import Any, TypeVar, TypeVarTuple +from typing import Any from homeassistant.core import ServiceResponse import homeassistant.util.dt as dt_util from .typing import TemplateVarsType -_T = TypeVar("_T") -_Ts = TypeVarTuple("_Ts") - class TraceElement: """Container for trace data.""" @@ -135,7 +132,9 @@ def trace_id_get() -> tuple[str, str] | None: return trace_id_cv.get() -def trace_stack_push(trace_stack_var: ContextVar[list[_T] | None], node: _T) -> None: +def trace_stack_push[_T]( + trace_stack_var: ContextVar[list[_T] | None], node: _T +) -> None: """Push an element to the top of a trace stack.""" trace_stack: list[_T] | None if (trace_stack := trace_stack_var.get()) is None: @@ -151,7 +150,7 @@ def trace_stack_pop(trace_stack_var: ContextVar[list[Any] | None]) -> None: trace_stack.pop() -def trace_stack_top(trace_stack_var: ContextVar[list[_T] | None]) -> _T | None: +def trace_stack_top[_T](trace_stack_var: ContextVar[list[_T] | None]) -> _T | None: """Return the element at the top of a trace stack.""" trace_stack = trace_stack_var.get() return trace_stack[-1] if trace_stack else None @@ -261,7 +260,7 @@ def trace_path(suffix: str | list[str]) -> Generator[None, None, None]: trace_path_pop(count) -def async_trace_path( +def async_trace_path[*_Ts]( suffix: str | list[str], ) -> Callable[ [Callable[[*_Ts], Coroutine[Any, Any, None]]],