Add integration filtering and error if core options to helpers.frame.report (#60009)

Reduces some existing and upcoming boilerplate.
This commit is contained in:
Ville Skyttä 2021-11-20 12:53:04 +02:00 committed by GitHub
parent e5c33474e3
commit f305d99af9
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 22 additions and 67 deletions

View File

@ -2,19 +2,11 @@
from __future__ import annotations
import asyncio
import contextlib
import logging
from typing import Any
import async_timeout
from homeassistant.helpers.frame import (
MissingIntegrationFrame,
get_integration_frame,
report_integration,
)
_LOGGER = logging.getLogger(__name__)
from homeassistant.helpers.frame import report
def timeout(
@ -24,8 +16,9 @@ def timeout(
if loop is None:
loop = asyncio.get_running_loop()
else:
_report(
"called async_timeout.timeout with loop keyword argument. The loop keyword argument is deprecated and calls will fail after Home Assistant 2022.2"
report(
"called async_timeout.timeout with loop keyword argument. The loop keyword argument is deprecated and calls will fail after Home Assistant 2022.2",
error_if_core=False,
)
if delay is not None:
deadline: float | None = loop.time() + delay
@ -36,8 +29,9 @@ def timeout(
def current_task(loop: asyncio.AbstractEventLoop) -> asyncio.Task[Any] | None:
"""Backwards compatible current_task."""
_report(
"called async_timeout.current_task. The current_task call is deprecated and calls will fail after Home Assistant 2022.2; use asyncio.current_task instead"
report(
"called async_timeout.current_task. The current_task call is deprecated and calls will fail after Home Assistant 2022.2; use asyncio.current_task instead",
error_if_core=False,
)
return asyncio.current_task()
@ -46,22 +40,3 @@ def enable() -> None:
"""Enable backwards compat transitions."""
async_timeout.timeout = timeout
async_timeout.current_task = current_task # type: ignore[attr-defined]
def _report(what: str) -> None:
"""Report incorrect usage.
Async friendly.
"""
integration_frame = None
with contextlib.suppress(MissingIntegrationFrame):
integration_frame = get_integration_frame()
if not integration_frame:
_LOGGER.warning(
"Detected code that %s; Please report this issue", what, stack_info=True
)
return
report_integration(what, integration_frame)

View File

@ -1,28 +1,22 @@
"""Zeroconf usage utility to warn about multiple instances."""
from contextlib import suppress
import logging
from typing import Any
import zeroconf
from homeassistant.helpers.frame import (
MissingIntegrationFrame,
get_integration_frame,
report_integration,
)
from homeassistant.helpers.frame import report
from .models import HaZeroconf
_LOGGER = logging.getLogger(__name__)
def install_multiple_zeroconf_catcher(hass_zc: HaZeroconf) -> None:
"""Wrap the Zeroconf class to return the shared instance if multiple instances are detected."""
def new_zeroconf_new(self: zeroconf.Zeroconf, *k: Any, **kw: Any) -> HaZeroconf:
_report(
report(
"attempted to create another Zeroconf instance. Please use the shared Zeroconf via await homeassistant.components.zeroconf.async_get_instance(hass)",
exclude_integrations={"zeroconf"},
error_if_core=False,
)
return hass_zc
@ -31,22 +25,3 @@ def install_multiple_zeroconf_catcher(hass_zc: HaZeroconf) -> None:
zeroconf.Zeroconf.__new__ = new_zeroconf_new # type: ignore
zeroconf.Zeroconf.__init__ = new_zeroconf_init # type: ignore
def _report(what: str) -> None:
"""Report incorrect usage.
Async friendly.
"""
integration_frame = None
with suppress(MissingIntegrationFrame):
integration_frame = get_integration_frame(exclude_integrations={"zeroconf"})
if not integration_frame:
_LOGGER.warning(
"Detected code that %s; Please report this issue", what, stack_info=True
)
return
report_integration(what, integration_frame)

View File

@ -50,18 +50,23 @@ class MissingIntegrationFrame(HomeAssistantError):
"""Raised when no integration is found in the frame."""
def report(what: str) -> None:
def report(
what: str, exclude_integrations: set | None = None, error_if_core: bool = True
) -> None:
"""Report incorrect usage.
Async friendly.
"""
try:
integration_frame = get_integration_frame()
integration_frame = get_integration_frame(
exclude_integrations=exclude_integrations
)
except MissingIntegrationFrame as err:
# Did not source from an integration? Hard error.
raise RuntimeError(
f"Detected code that {what}. Please report this issue."
) from err
msg = f"Detected code that {what}. Please report this issue."
if error_if_core:
raise RuntimeError(msg) from err
_LOGGER.warning(msg, stack_info=True)
return
report_integration(what, integration_frame)