Allow importing typing helper in core files (#119377)

* Allow importing typing helper in core files

* Really fix the circular import

* Update test
pull/119393/head
Erik Montnemery 2024-06-11 13:48:12 +02:00 committed by GitHub
parent 572700a326
commit 904b89df80
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 69 additions and 44 deletions

View File

@ -96,6 +96,7 @@ from .helpers.deprecation import (
dir_with_deprecated_constants,
)
from .helpers.json import json_bytes, json_fragment
from .helpers.typing import UNDEFINED, UndefinedType
from .util import dt as dt_util, location
from .util.async_ import (
cancelling,
@ -131,8 +132,6 @@ FINAL_WRITE_STAGE_SHUTDOWN_TIMEOUT = 60
CLOSE_STAGE_SHUTDOWN_TIMEOUT = 30
# Internal; not helpers.typing.UNDEFINED due to circular dependency
_UNDEF: dict[Any, Any] = {}
_SENTINEL = object()
_DataT = TypeVar("_DataT", bound=Mapping[str, Any], default=Mapping[str, Any])
type CALLBACK_TYPE = Callable[[], None]
@ -3035,11 +3034,10 @@ class Config:
unit_system: str | None = None,
location_name: str | None = None,
time_zone: str | None = None,
# pylint: disable=dangerous-default-value # _UNDEFs not modified
external_url: str | dict[Any, Any] | None = _UNDEF,
internal_url: str | dict[Any, Any] | None = _UNDEF,
external_url: str | UndefinedType | None = UNDEFINED,
internal_url: str | UndefinedType | None = UNDEFINED,
currency: str | None = None,
country: str | dict[Any, Any] | None = _UNDEF,
country: str | UndefinedType | None = UNDEFINED,
language: str | None = None,
) -> None:
"""Update the configuration from a dictionary."""
@ -3059,14 +3057,14 @@ class Config:
self.location_name = location_name
if time_zone is not None:
await self.async_set_time_zone(time_zone)
if external_url is not _UNDEF:
self.external_url = cast(str | None, external_url)
if internal_url is not _UNDEF:
self.internal_url = cast(str | None, internal_url)
if external_url is not UNDEFINED:
self.external_url = external_url
if internal_url is not UNDEFINED:
self.internal_url = internal_url
if currency is not None:
self.currency = currency
if country is not _UNDEF:
self.country = cast(str | None, country)
if country is not UNDEFINED:
self.country = country
if language is not None:
self.language = language
@ -3112,8 +3110,8 @@ class Config:
unit_system=data.get("unit_system_v2"),
location_name=data.get("location_name"),
time_zone=data.get("time_zone"),
external_url=data.get("external_url", _UNDEF),
internal_url=data.get("internal_url", _UNDEF),
external_url=data.get("external_url", UNDEFINED),
internal_url=data.get("internal_url", UNDEFINED),
currency=data.get("currency"),
country=data.get("country"),
language=data.get("language"),

View File

@ -242,6 +242,26 @@ class DeprecatedAlias(NamedTuple):
breaks_in_ha_version: str | None
class DeferredDeprecatedAlias:
"""Deprecated alias with deferred evaluation of the value."""
def __init__(
self,
value_fn: Callable[[], Any],
replacement: str,
breaks_in_ha_version: str | None,
) -> None:
"""Initialize."""
self.breaks_in_ha_version = breaks_in_ha_version
self.replacement = replacement
self._value_fn = value_fn
@functools.cached_property
def value(self) -> Any:
"""Return the value."""
return self._value_fn()
_PREFIX_DEPRECATED = "_DEPRECATED_"
@ -266,7 +286,7 @@ def check_if_deprecated_constant(name: str, module_globals: dict[str, Any]) -> A
f"{deprecated_const.enum.__class__.__name__}.{deprecated_const.enum.name}"
)
breaks_in_ha_version = deprecated_const.breaks_in_ha_version
elif isinstance(deprecated_const, DeprecatedAlias):
elif isinstance(deprecated_const, (DeprecatedAlias, DeferredDeprecatedAlias)):
description = "alias"
value = deprecated_const.value
replacement = deprecated_const.replacement
@ -274,8 +294,10 @@ def check_if_deprecated_constant(name: str, module_globals: dict[str, Any]) -> A
if value is None or replacement is None:
msg = (
f"Value of {_PREFIX_DEPRECATED}{name} is an instance of {type(deprecated_const)} "
"but an instance of DeprecatedConstant or DeprecatedConstantEnum is required"
f"Value of {_PREFIX_DEPRECATED}{name} is an instance of "
f"{type(deprecated_const)} but an instance of DeprecatedAlias, "
"DeferredDeprecatedAlias, DeprecatedConstant or DeprecatedConstantEnum "
"is required"
)
logging.getLogger(module_name).debug(msg)

View File

@ -5,10 +5,8 @@ from enum import Enum
from functools import partial
from typing import Any, Never
import homeassistant.core
from .deprecation import (
DeprecatedAlias,
DeferredDeprecatedAlias,
all_with_deprecated_constants,
check_if_deprecated_constant,
dir_with_deprecated_constants,
@ -35,23 +33,27 @@ class UndefinedType(Enum):
UNDEFINED = UndefinedType._singleton # noqa: SLF001
def _deprecated_typing_helper(attr: str) -> DeferredDeprecatedAlias:
"""Help to make a DeferredDeprecatedAlias."""
def value_fn() -> Any:
# pylint: disable-next=import-outside-toplevel
import homeassistant.core
return getattr(homeassistant.core, attr)
return DeferredDeprecatedAlias(value_fn, f"homeassistant.core.{attr}", "2025.5")
# The following types should not used and
# are not present in the core code base.
# They are kept in order not to break custom integrations
# that may rely on them.
# Deprecated as of 2024.5 use types from homeassistant.core instead.
_DEPRECATED_ContextType = DeprecatedAlias(
homeassistant.core.Context, "homeassistant.core.Context", "2025.5"
)
_DEPRECATED_EventType = DeprecatedAlias(
homeassistant.core.Event, "homeassistant.core.Event", "2025.5"
)
_DEPRECATED_HomeAssistantType = DeprecatedAlias(
homeassistant.core.HomeAssistant, "homeassistant.core.HomeAssistant", "2025.5"
)
_DEPRECATED_ServiceCallType = DeprecatedAlias(
homeassistant.core.ServiceCall, "homeassistant.core.ServiceCall", "2025.5"
)
_DEPRECATED_ContextType = _deprecated_typing_helper("Context")
_DEPRECATED_EventType = _deprecated_typing_helper("Event")
_DEPRECATED_HomeAssistantType = _deprecated_typing_helper("HomeAssistant")
_DEPRECATED_ServiceCallType = _deprecated_typing_helper("ServiceCall")
# These can be removed if no deprecated constant are in this module anymore
__getattr__ = partial(check_if_deprecated_constant, module_globals=globals())

View File

@ -40,6 +40,7 @@ from .generated.ssdp import SSDP
from .generated.usb import USB
from .generated.zeroconf import HOMEKIT, ZEROCONF
from .helpers.json import json_bytes, json_fragment
from .helpers.typing import UNDEFINED
from .util.hass_dict import HassKey
from .util.json import JSON_DECODE_EXCEPTIONS, json_loads
@ -129,9 +130,6 @@ IMPORT_EVENT_LOOP_WARNING = (
"experience issues with Home Assistant"
)
_UNDEF = object() # Internal; not helpers.typing.UNDEFINED due to circular dependency
MOVED_ZEROCONF_PROPS = ("macaddress", "model", "manufacturer")
@ -1322,7 +1320,7 @@ def async_get_loaded_integration(hass: HomeAssistant, domain: str) -> Integratio
Raises IntegrationNotLoaded if the integration is not loaded.
"""
cache = hass.data[DATA_INTEGRATIONS]
int_or_fut = cache.get(domain, _UNDEF)
int_or_fut = cache.get(domain, UNDEFINED)
# Integration is never subclassed, so we can check for type
if type(int_or_fut) is Integration:
return int_or_fut
@ -1332,7 +1330,7 @@ def async_get_loaded_integration(hass: HomeAssistant, domain: str) -> Integratio
async def async_get_integration(hass: HomeAssistant, domain: str) -> Integration:
"""Get integration."""
cache = hass.data[DATA_INTEGRATIONS]
if type(int_or_fut := cache.get(domain, _UNDEF)) is Integration:
if type(int_or_fut := cache.get(domain, UNDEFINED)) is Integration:
return int_or_fut
integrations_or_excs = await async_get_integrations(hass, [domain])
int_or_exc = integrations_or_excs[domain]
@ -1350,11 +1348,11 @@ async def async_get_integrations(
needed: dict[str, asyncio.Future[None]] = {}
in_progress: dict[str, asyncio.Future[None]] = {}
for domain in domains:
int_or_fut = cache.get(domain, _UNDEF)
int_or_fut = cache.get(domain, UNDEFINED)
# Integration is never subclassed, so we can check for type
if type(int_or_fut) is Integration:
results[domain] = int_or_fut
elif int_or_fut is not _UNDEF:
elif int_or_fut is not UNDEFINED:
in_progress[domain] = cast(asyncio.Future[None], int_or_fut)
elif "." in domain:
results[domain] = ValueError(f"Invalid domain {domain}")
@ -1364,10 +1362,10 @@ async def async_get_integrations(
if in_progress:
await asyncio.wait(in_progress.values())
for domain in in_progress:
# When we have waited and it's _UNDEF, it doesn't exist
# When we have waited and it's UNDEFINED, it doesn't exist
# We don't cache that it doesn't exist, or else people can't fix it
# and then restart, because their config will never be valid.
if (int_or_fut := cache.get(domain, _UNDEF)) is _UNDEF:
if (int_or_fut := cache.get(domain, UNDEFINED)) is UNDEFINED:
results[domain] = IntegrationNotFound(domain)
else:
results[domain] = cast(Integration, int_or_fut)

View File

@ -483,14 +483,19 @@ def test_check_if_deprecated_constant_integration_not_found(
def test_test_check_if_deprecated_constant_invalid(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Test check_if_deprecated_constant will raise an attribute error and create an log entry on an invalid deprecation type."""
"""Test check_if_deprecated_constant error handling.
Test check_if_deprecated_constant raises an attribute error and creates a log entry
on an invalid deprecation type.
"""
module_name = "homeassistant.components.hue.light"
module_globals = {"__name__": module_name, "_DEPRECATED_TEST_CONSTANT": 1}
name = "TEST_CONSTANT"
excepted_msg = (
f"Value of _DEPRECATED_{name} is an instance of <class 'int'> "
"but an instance of DeprecatedConstant or DeprecatedConstantEnum is required"
f"Value of _DEPRECATED_{name} is an instance of <class 'int'> but an instance "
"of DeprecatedAlias, DeferredDeprecatedAlias, DeprecatedConstant or "
"DeprecatedConstantEnum is required"
)
with pytest.raises(AttributeError, match=excepted_msg):