Make TypeVars private (1) (#68205)
parent
a9c383a1e5
commit
be7ef6115c
|
@ -8,7 +8,7 @@ from homeassistant.core import callback
|
|||
|
||||
from .const import REDACTED
|
||||
|
||||
T = TypeVar("T")
|
||||
_T = TypeVar("_T")
|
||||
|
||||
|
||||
@overload
|
||||
|
@ -17,18 +17,18 @@ 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(data: _T, to_redact: Iterable[Any]) -> _T:
|
||||
...
|
||||
|
||||
|
||||
@callback
|
||||
def async_redact_data(data: T, to_redact: Iterable[Any]) -> T:
|
||||
def async_redact_data(data: _T, to_redact: Iterable[Any]) -> _T:
|
||||
"""Redact sensitive data in a dict."""
|
||||
if not isinstance(data, (Mapping, list)):
|
||||
return data
|
||||
|
||||
if isinstance(data, list):
|
||||
return cast(T, [async_redact_data(val, to_redact) for val in data])
|
||||
return cast(_T, [async_redact_data(val, to_redact) for val in data])
|
||||
|
||||
redacted = {**data}
|
||||
|
||||
|
@ -40,4 +40,4 @@ def async_redact_data(data: T, to_redact: Iterable[Any]) -> T:
|
|||
elif isinstance(value, list):
|
||||
redacted[key] = [async_redact_data(item, to_redact) for item in value]
|
||||
|
||||
return cast(T, redacted)
|
||||
return cast(_T, redacted)
|
||||
|
|
|
@ -99,15 +99,13 @@ STAGE_3_SHUTDOWN_TIMEOUT = 30
|
|||
|
||||
block_async_io.enable()
|
||||
|
||||
T = TypeVar("T")
|
||||
_T = TypeVar("_T")
|
||||
_R = TypeVar("_R")
|
||||
_R_co = TypeVar("_R_co", covariant=True) # pylint: disable=invalid-name
|
||||
# Internal; not helpers.typing.UNDEFINED due to circular dependency
|
||||
_UNDEF: dict[Any, Any] = {}
|
||||
# pylint: disable=invalid-name
|
||||
CALLABLE_T = TypeVar("CALLABLE_T", bound=Callable[..., Any])
|
||||
CALLBACK_TYPE = Callable[[], None]
|
||||
# pylint: enable=invalid-name
|
||||
_CallableT = TypeVar("_CallableT", bound=Callable[..., Any])
|
||||
CALLBACK_TYPE = Callable[[], None] # pylint: disable=invalid-name
|
||||
|
||||
CORE_STORAGE_KEY = "core.config"
|
||||
CORE_STORAGE_VERSION = 1
|
||||
|
@ -165,7 +163,7 @@ def valid_state(state: str) -> bool:
|
|||
return len(state) <= MAX_LENGTH_STATE_STATE
|
||||
|
||||
|
||||
def callback(func: CALLABLE_T) -> CALLABLE_T:
|
||||
def callback(func: _CallableT) -> _CallableT:
|
||||
"""Annotation to mark method as safe to call from within the event loop."""
|
||||
setattr(func, "_hass_callback", True)
|
||||
return func
|
||||
|
@ -480,8 +478,8 @@ class HomeAssistant:
|
|||
|
||||
@callback
|
||||
def async_add_executor_job(
|
||||
self, target: Callable[..., T], *args: Any
|
||||
) -> asyncio.Future[T]:
|
||||
self, target: Callable[..., _T], *args: Any
|
||||
) -> asyncio.Future[_T]:
|
||||
"""Add an executor job from within the event loop."""
|
||||
task = self.loop.run_in_executor(None, target, *args)
|
||||
|
||||
|
|
|
@ -35,9 +35,7 @@ from .util.async_ import gather_with_concurrency
|
|||
if TYPE_CHECKING:
|
||||
from .core import HomeAssistant
|
||||
|
||||
CALLABLE_T = TypeVar( # pylint: disable=invalid-name
|
||||
"CALLABLE_T", bound=Callable[..., Any]
|
||||
)
|
||||
_CallableT = TypeVar("_CallableT", bound=Callable[..., Any])
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
@ -805,7 +803,7 @@ class Helpers:
|
|||
return wrapped
|
||||
|
||||
|
||||
def bind_hass(func: CALLABLE_T) -> CALLABLE_T:
|
||||
def bind_hass(func: _CallableT) -> _CallableT:
|
||||
"""Decorate function to indicate that first argument is hass."""
|
||||
setattr(func, "__bind_hass", True)
|
||||
return func
|
||||
|
|
|
@ -22,7 +22,7 @@ from homeassistant.util import dt as dt_util
|
|||
# mypy: allow-untyped-calls, allow-untyped-defs, no-check-untyped-defs
|
||||
# mypy: no-warn-return-any
|
||||
|
||||
CALLABLE_T = TypeVar("CALLABLE_T", bound=Callable) # pylint: disable=invalid-name
|
||||
_CallableT = TypeVar("_CallableT", bound=Callable)
|
||||
|
||||
BENCHMARKS: dict[str, Callable] = {}
|
||||
|
||||
|
@ -54,7 +54,7 @@ async def run_benchmark(bench):
|
|||
await hass.async_stop()
|
||||
|
||||
|
||||
def benchmark(func: CALLABLE_T) -> CALLABLE_T:
|
||||
def benchmark(func: _CallableT) -> _CallableT:
|
||||
"""Decorate to mark a benchmark."""
|
||||
BENCHMARKS[func.__name__] = func
|
||||
return func
|
||||
|
|
|
@ -15,8 +15,8 @@ import slugify as unicode_slug
|
|||
|
||||
from .dt import as_local, utcnow
|
||||
|
||||
T = TypeVar("T")
|
||||
U = TypeVar("U") # pylint: disable=invalid-name
|
||||
_T = TypeVar("_T")
|
||||
_U = TypeVar("_U")
|
||||
|
||||
RE_SANITIZE_FILENAME = re.compile(r"(~|\.\.|/|\\)")
|
||||
RE_SANITIZE_PATH = re.compile(r"(~|\.(\.)+)")
|
||||
|
@ -63,8 +63,8 @@ def repr_helper(inp: Any) -> str:
|
|||
|
||||
|
||||
def convert(
|
||||
value: T | None, to_type: Callable[[T], U], default: U | None = None
|
||||
) -> U | None:
|
||||
value: _T | None, to_type: Callable[[_T], _U], default: _U | None = None
|
||||
) -> _U | None:
|
||||
"""Convert value to to_type, returns default if fails."""
|
||||
try:
|
||||
return default if value is None else to_type(value)
|
||||
|
|
|
@ -3,10 +3,10 @@ from __future__ import annotations
|
|||
|
||||
from typing import TypeVar
|
||||
|
||||
T = TypeVar("T")
|
||||
_T = TypeVar("_T")
|
||||
|
||||
|
||||
def ordered_list_item_to_percentage(ordered_list: list[T], item: T) -> int:
|
||||
def ordered_list_item_to_percentage(ordered_list: list[_T], item: _T) -> int:
|
||||
"""Determine the percentage of an item in an ordered list.
|
||||
|
||||
When using this utility for fan speeds, do not include "off"
|
||||
|
@ -29,7 +29,7 @@ def ordered_list_item_to_percentage(ordered_list: list[T], item: T) -> int:
|
|||
return (list_position * 100) // list_len
|
||||
|
||||
|
||||
def percentage_to_ordered_list_item(ordered_list: list[T], percentage: int) -> T:
|
||||
def percentage_to_ordered_list_item(ordered_list: list[_T], percentage: int) -> _T:
|
||||
"""Find the item that most closely matches the percentage in an ordered list.
|
||||
|
||||
When using this utility for fan speeds, do not include "off"
|
||||
|
|
|
@ -7,11 +7,11 @@ def _readonly(*args: Any, **kwargs: Any) -> Any:
|
|||
raise RuntimeError("Cannot modify ReadOnlyDict")
|
||||
|
||||
|
||||
Key = TypeVar("Key")
|
||||
Value = TypeVar("Value")
|
||||
_KT = TypeVar("_KT")
|
||||
_VT = TypeVar("_VT")
|
||||
|
||||
|
||||
class ReadOnlyDict(dict[Key, Value]):
|
||||
class ReadOnlyDict(dict[_KT, _VT]):
|
||||
"""Read only version of dict that is compatible with dict types."""
|
||||
|
||||
__setitem__ = _readonly
|
||||
|
|
|
@ -19,7 +19,7 @@ from .objects import Input, NodeListClass, NodeStrClass
|
|||
# mypy: allow-untyped-calls, no-warn-return-any
|
||||
|
||||
JSON_TYPE = Union[list, dict, str] # pylint: disable=invalid-name
|
||||
DICT_T = TypeVar("DICT_T", bound=dict) # pylint: disable=invalid-name
|
||||
_DictT = TypeVar("_DictT", bound=dict)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
@ -144,8 +144,8 @@ def _add_reference(
|
|||
|
||||
@overload
|
||||
def _add_reference(
|
||||
obj: DICT_T, loader: SafeLineLoader, node: yaml.nodes.Node
|
||||
) -> DICT_T:
|
||||
obj: _DictT, loader: SafeLineLoader, node: yaml.nodes.Node
|
||||
) -> _DictT:
|
||||
...
|
||||
|
||||
|
||||
|
|
Loading…
Reference in New Issue