Use PEP 695 for function annotations (1) (#117658)

This commit is contained in:
Marc Mueller 2024-05-18 11:45:54 +02:00 committed by GitHub
parent 900b6211ef
commit 26a599ad11
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 43 additions and 89 deletions

View file

@ -7,7 +7,7 @@ from collections.abc import Callable, Coroutine, Iterable, ValuesView
from datetime import datetime
from itertools import chain
import logging
from typing import Any, ParamSpec, TypeVar
from typing import Any
from aiohttp import ClientError, ClientResponseError
from yalexs.activity import ActivityTypes
@ -36,9 +36,6 @@ from .gateway import AugustGateway
from .subscriber import AugustSubscriberMixin
from .util import async_create_august_clientsession
_R = TypeVar("_R")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
API_CACHED_ATTRS = {
@ -403,7 +400,7 @@ class AugustData(AugustSubscriberMixin):
hyper_bridge,
)
async def _async_call_api_op_requires_bridge(
async def _async_call_api_op_requires_bridge[**_P, _R](
self,
device_id: str,
func: Callable[_P, Coroutine[Any, Any, _R]],

View file

@ -3,7 +3,7 @@
from __future__ import annotations
import logging
from typing import Any, Self, TypeVar
from typing import Any, Self
import voluptuous as vol
@ -23,8 +23,6 @@ from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.storage import Store
from homeassistant.helpers.typing import ConfigType
_T = TypeVar("_T")
_LOGGER = logging.getLogger(__name__)
ATTR_INITIAL = "initial"
@ -62,7 +60,7 @@ STORAGE_FIELDS = {
}
def _none_to_empty_dict(value: _T | None) -> _T | dict[str, Any]:
def _none_to_empty_dict[_T](value: _T | None) -> _T | dict[str, Any]:
if value is None:
return {}
return value

View file

@ -8,7 +8,7 @@ from enum import IntFlag, StrEnum
import functools as ft
from functools import cached_property
import logging
from typing import Any, ParamSpec, TypeVar, final
from typing import Any, final
import voluptuous as vol
@ -54,9 +54,6 @@ SCAN_INTERVAL = timedelta(seconds=15)
ENTITY_ID_FORMAT = DOMAIN + ".{}"
_P = ParamSpec("_P")
_R = TypeVar("_R")
class CoverDeviceClass(StrEnum):
"""Device class for cover."""
@ -477,7 +474,7 @@ class CoverEntity(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_):
else:
await self.async_close_cover_tilt(**kwargs)
def _get_toggle_function(
def _get_toggle_function[**_P, _R](
self, fns: dict[str, Callable[_P, _R]]
) -> Callable[_P, _R]:
# If we are opening or closing and we support stopping, then we should stop

View file

@ -3,14 +3,12 @@
from __future__ import annotations
from collections.abc import Iterable, Mapping
from typing import Any, TypeVar, cast, overload
from typing import Any, cast, overload
from homeassistant.core import callback
from .const import REDACTED
_T = TypeVar("_T")
@overload
def async_redact_data(data: Mapping, to_redact: Iterable[Any]) -> dict: # type: ignore[overload-overlap]
@ -18,11 +16,11 @@ def async_redact_data(data: Mapping, to_redact: Iterable[Any]) -> dict: # type:
@overload
def async_redact_data(data: _T, to_redact: Iterable[Any]) -> _T: ...
def async_redact_data[_T](data: _T, to_redact: Iterable[Any]) -> _T: ...
@callback
def async_redact_data(data: _T, to_redact: Iterable[Any]) -> _T:
def async_redact_data[_T](data: _T, to_redact: Iterable[Any]) -> _T:
"""Redact sensitive data in a dict."""
if not isinstance(data, (Mapping, list)):
return data

View file

@ -3,7 +3,7 @@
from abc import ABC, abstractmethod
from collections.abc import Callable
import logging
from typing import Any, TypeVar, cast
from typing import Any, cast
from fitbit import Fitbit
from fitbit.exceptions import HTTPException, HTTPUnauthorized
@ -24,9 +24,6 @@ CONF_REFRESH_TOKEN = "refresh_token"
CONF_EXPIRES_AT = "expires_at"
_T = TypeVar("_T")
class FitbitApi(ABC):
"""Fitbit client library wrapper base class.
@ -129,7 +126,7 @@ class FitbitApi(ABC):
dated_results: list[dict[str, Any]] = response[key]
return dated_results[-1]
async def _run(self, func: Callable[[], _T]) -> _T:
async def _run[_T](self, func: Callable[[], _T]) -> _T:
"""Run client command."""
try:
return await self._hass.async_add_executor_job(func)

View file

@ -5,7 +5,7 @@ from __future__ import annotations
import asyncio
from datetime import datetime, timedelta
import logging
from typing import Final, TypeVar
from typing import Final
from pyfronius import Fronius, FroniusError
@ -39,8 +39,6 @@ from .coordinator import (
_LOGGER: Final = logging.getLogger(__name__)
PLATFORMS: Final = [Platform.SENSOR]
_FroniusCoordinatorT = TypeVar("_FroniusCoordinatorT", bound=FroniusCoordinatorBase)
type FroniusConfigEntry = ConfigEntry[FroniusSolarNet]
@ -255,7 +253,7 @@ class FroniusSolarNet:
return inverter_infos
@staticmethod
async def _init_optional_coordinator(
async def _init_optional_coordinator[_FroniusCoordinatorT: FroniusCoordinatorBase](
coordinator: _FroniusCoordinatorT,
) -> _FroniusCoordinatorT | None:
"""Initialize an update coordinator and return it if devices are found."""

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from datetime import datetime, timedelta
import logging
from typing import Any, TypeVar
from typing import Any
from homeassistant.components import (
alarm_control_panel,
@ -242,10 +242,8 @@ COVER_VALVE_DOMAINS = {cover.DOMAIN, valve.DOMAIN}
FRIENDLY_DOMAIN = {cover.DOMAIN: "Cover", valve.DOMAIN: "Valve"}
_TraitT = TypeVar("_TraitT", bound="_Trait")
def register_trait(trait: type[_TraitT]) -> type[_TraitT]:
def register_trait[_TraitT: _Trait](trait: type[_TraitT]) -> type[_TraitT]:
"""Decorate a class to register a trait."""
TRAITS.append(trait)
return trait

View file

@ -4,7 +4,7 @@ from __future__ import annotations
from abc import abstractmethod
import datetime
from typing import Any, TypeVar
from typing import Any
import voluptuous as vol
@ -55,10 +55,8 @@ UNITS: dict[str, str] = {
}
ICON = "mdi:chart-line"
_T = TypeVar("_T", bound=dict[str, Any])
def exactly_two_period_keys(conf: _T) -> _T:
def exactly_two_period_keys[_T: dict[str, Any]](conf: _T) -> _T:
"""Ensure exactly 2 of CONF_PERIOD_KEYS are provided."""
if sum(param in conf for param in CONF_PERIOD_KEYS) != 2:
raise vol.Invalid(

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable
from datetime import timedelta
import logging
from typing import Any, TypeVar
from typing import Any
import voluptuous as vol
@ -41,10 +41,8 @@ from homeassistant.helpers.event import (
from homeassistant.helpers.trigger import TriggerActionType, TriggerInfo
from homeassistant.helpers.typing import ConfigType
_T = TypeVar("_T", bound=dict[str, Any])
def validate_above_below(value: _T) -> _T:
def validate_above_below[_T: dict[str, Any]](value: _T) -> _T:
"""Validate that above and below can co-exist."""
above = value.get(CONF_ABOVE)
below = value.get(CONF_BELOW)

View file

@ -6,7 +6,7 @@ import asyncio
from collections.abc import Callable, Coroutine
from dataclasses import dataclass
import logging
from typing import Any, TypeVar
from typing import Any
from bleak import BleakError
from improv_ble_client import (
@ -30,8 +30,6 @@ from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
_T = TypeVar("_T")
STEP_PROVISION_SCHEMA = vol.Schema(
{
vol.Required("ssid"): str,
@ -392,7 +390,7 @@ class ImprovBLEConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_show_progress_done(next_step_id="provision")
@staticmethod
async def _try_call(func: Coroutine[Any, Any, _T]) -> _T:
async def _try_call[_T](func: Coroutine[Any, Any, _T]) -> _T:
"""Call the library and abort flow on common errors."""
try:
return await func

View file

@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import Any, TypeVar
from typing import Any
from linear_garage_door import Linear
from linear_garage_door.errors import InvalidLoginError
@ -19,8 +19,6 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
_T = TypeVar("_T")
@dataclass
class LinearDevice:
@ -63,7 +61,7 @@ class LinearUpdateCoordinator(DataUpdateCoordinator[dict[str, LinearDevice]]):
return await self.execute(update_data)
async def execute(self, func: Callable[[Linear], Awaitable[_T]]) -> _T:
async def execute[_T](self, func: Callable[[Linear], Awaitable[_T]]) -> _T:
"""Execute an API call."""
linear = Linear()
try:

View file

@ -1,7 +1,6 @@
"""Melnor integration models."""
from collections.abc import Callable
from typing import TypeVar
from melnor_bluetooth.device import Device, Valve
@ -77,14 +76,11 @@ class MelnorZoneEntity(MelnorBluetoothEntity):
)
T = TypeVar("T", bound=EntityDescription)
def get_entities_for_valves(
def get_entities_for_valves[_T: EntityDescription](
coordinator: MelnorDataUpdateCoordinator,
descriptions: list[T],
descriptions: list[_T],
function: Callable[
[Valve, T],
[Valve, _T],
CoordinatorEntity[MelnorDataUpdateCoordinator],
],
) -> list[CoordinatorEntity[MelnorDataUpdateCoordinator]]:

View file

@ -3,7 +3,7 @@
from __future__ import annotations
from collections.abc import Coroutine
from typing import Any, TypeVar
from typing import Any
from urllib.error import URLError
from radiotherm.validate import RadiothermTstatError
@ -20,10 +20,8 @@ from .util import async_set_time
PLATFORMS: list[Platform] = [Platform.CLIMATE, Platform.SWITCH]
_T = TypeVar("_T")
async def _async_call_or_raise_not_ready(
async def _async_call_or_raise_not_ready[_T](
coro: Coroutine[Any, Any, _T], host: str
) -> _T:
"""Call a coro or raise ConfigEntryNotReady."""

View file

@ -12,7 +12,7 @@ import queue
import sqlite3
import threading
import time
from typing import TYPE_CHECKING, Any, TypeVar, cast
from typing import TYPE_CHECKING, Any, cast
import psutil_home_assistant as ha_psutil
from sqlalchemy import create_engine, event as sqlalchemy_event, exc, select, update
@ -138,8 +138,6 @@ from .util import (
_LOGGER = logging.getLogger(__name__)
T = TypeVar("T")
DEFAULT_URL = "sqlite:///{hass_config_path}"
# Controls how often we clean up
@ -366,9 +364,9 @@ class Recorder(threading.Thread):
self.queue_task(COMMIT_TASK)
@callback
def async_add_executor_job(
self, target: Callable[..., T], *args: Any
) -> asyncio.Future[T]:
def async_add_executor_job[_T](
self, target: Callable[..., _T], *args: Any
) -> asyncio.Future[_T]:
"""Add an executor job from within the event loop."""
return self.hass.loop.run_in_executor(self._db_executor, target, *args)

View file

@ -6,7 +6,7 @@ import binascii
from collections.abc import Callable, Mapping
import copy
import logging
from typing import Any, NamedTuple, TypeVarTuple, cast
from typing import Any, NamedTuple, cast
import RFXtrx as rfxtrxmod
import voluptuous as vol
@ -55,8 +55,6 @@ DEFAULT_OFF_DELAY = 2.0
SIGNAL_EVENT = f"{DOMAIN}_event"
CONNECT_TIMEOUT = 30.0
_Ts = TypeVarTuple("_Ts")
_LOGGER = logging.getLogger(__name__)
@ -573,7 +571,7 @@ class RfxtrxCommandEntity(RfxtrxEntity):
"""Initialzie a switch or light device."""
super().__init__(device, device_id, event=event)
async def _async_send(
async def _async_send[*_Ts](
self, fun: Callable[[rfxtrxmod.PySerialTransport, *_Ts], None], *args: *_Ts
) -> None:
rfx_object: rfxtrxmod.Connect = self.hass.data[DOMAIN][DATA_RFXOBJECT]

View file

@ -3,7 +3,6 @@
from asyncio import TaskGroup
from collections.abc import Callable
import logging
from typing import TypeVar, TypeVarTuple
from ring_doorbell import AuthenticationError, Ring, RingDevices, RingError, RingTimeout
@ -15,11 +14,8 @@ from .const import NOTIFICATIONS_SCAN_INTERVAL, SCAN_INTERVAL
_LOGGER = logging.getLogger(__name__)
_R = TypeVar("_R")
_Ts = TypeVarTuple("_Ts")
async def _call_api(
async def _call_api[*_Ts, _R](
hass: HomeAssistant, target: Callable[[*_Ts], _R], *args: *_Ts, msg_suffix: str = ""
) -> _R:
try:

View file

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine
import logging
from typing import Any, TypeVar
from typing import Any
from api.soma_api import SomaApi
from requests import RequestException
@ -22,8 +22,6 @@ from homeassistant.helpers.typing import ConfigType
from .const import API, DOMAIN, HOST, PORT
from .utils import is_api_response_success
_SomaEntityT = TypeVar("_SomaEntityT", bound="SomaEntity")
_LOGGER = logging.getLogger(__name__)
DEVICES = "devices"
@ -76,7 +74,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
def soma_api_call(
def soma_api_call[_SomaEntityT: SomaEntity](
api_call: Callable[[_SomaEntityT], Coroutine[Any, Any, dict]],
) -> Callable[[_SomaEntityT], Coroutine[Any, Any, dict]]:
"""Soma api call decorator."""

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable
from datetime import datetime, timedelta
import logging
from typing import Any, Self, TypeVar
from typing import Any, Self
import voluptuous as vol
@ -29,7 +29,6 @@ from homeassistant.helpers.storage import Store
from homeassistant.helpers.typing import ConfigType
import homeassistant.util.dt as dt_util
_T = TypeVar("_T")
_LOGGER = logging.getLogger(__name__)
DOMAIN = "timer"
@ -82,7 +81,7 @@ def _format_timedelta(delta: timedelta) -> str:
return f"{int(hours)}:{int(minutes):02}:{int(seconds):02}"
def _none_to_empty_dict(value: _T | None) -> _T | dict[Any, Any]:
def _none_to_empty_dict[_T](value: _T | None) -> _T | dict[Any, Any]:
if value is None:
return {}
return value

View file

@ -14,7 +14,7 @@ from dataclasses import dataclass
import enum
import logging
import re
from typing import TYPE_CHECKING, Any, TypeVar, overload
from typing import TYPE_CHECKING, Any, overload
import voluptuous as vol
import zigpy.exceptions
@ -62,7 +62,6 @@ if TYPE_CHECKING:
from .device import ZHADevice
from .gateway import ZHAGateway
_T = TypeVar("_T")
_LOGGER = logging.getLogger(__name__)
@ -228,7 +227,7 @@ def async_is_bindable_target(source_zha_device, target_zha_device):
@callback
def async_get_zha_config_value(
def async_get_zha_config_value[_T](
config_entry: ConfigEntry, section: str, config_key: str, default: _T
) -> _T:
"""Get the value for the specified configuration from the ZHA config entry."""

View file

@ -4,7 +4,7 @@ from __future__ import annotations
import asyncio
import logging
from typing import TYPE_CHECKING, Any, Literal, NamedTuple, TypeVar, cast
from typing import TYPE_CHECKING, Any, Literal, NamedTuple, cast
import voluptuous as vol
import zigpy.backups
@ -118,11 +118,8 @@ IEEE_SERVICE = "ieee_based_service"
IEEE_SCHEMA = vol.All(cv.string, EUI64.convert)
# typing typevar
_T = TypeVar("_T")
def _ensure_list_if_present(value: _T | None) -> list[_T] | list[Any] | None:
def _ensure_list_if_present[_T](value: _T | None) -> list[_T] | list[Any] | None:
"""Wrap value in list if it is provided and not one."""
if value is None:
return None