mirror of
https://github.com/home-assistant/core.git
synced 2025-07-24 21:57:51 +00:00
Postponed evaluation of annotations in core (#46434)
* Postponed evaluation of annotations in core * Remove unneeded future
This commit is contained in:
parent
910c034613
commit
9b7c39d20b
@ -1,4 +1,6 @@
|
|||||||
"""Provide an authentication layer for Home Assistant."""
|
"""Provide an authentication layer for Home Assistant."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
@ -36,7 +38,7 @@ async def auth_manager_from_config(
|
|||||||
hass: HomeAssistant,
|
hass: HomeAssistant,
|
||||||
provider_configs: List[Dict[str, Any]],
|
provider_configs: List[Dict[str, Any]],
|
||||||
module_configs: List[Dict[str, Any]],
|
module_configs: List[Dict[str, Any]],
|
||||||
) -> "AuthManager":
|
) -> AuthManager:
|
||||||
"""Initialize an auth manager from config.
|
"""Initialize an auth manager from config.
|
||||||
|
|
||||||
CORE_CONFIG_SCHEMA will make sure do duplicated auth providers or
|
CORE_CONFIG_SCHEMA will make sure do duplicated auth providers or
|
||||||
@ -76,7 +78,7 @@ async def auth_manager_from_config(
|
|||||||
class AuthManagerFlowManager(data_entry_flow.FlowManager):
|
class AuthManagerFlowManager(data_entry_flow.FlowManager):
|
||||||
"""Manage authentication flows."""
|
"""Manage authentication flows."""
|
||||||
|
|
||||||
def __init__(self, hass: HomeAssistant, auth_manager: "AuthManager"):
|
def __init__(self, hass: HomeAssistant, auth_manager: AuthManager):
|
||||||
"""Init auth manager flows."""
|
"""Init auth manager flows."""
|
||||||
super().__init__(hass)
|
super().__init__(hass)
|
||||||
self.auth_manager = auth_manager
|
self.auth_manager = auth_manager
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Pluggable auth modules for Home Assistant."""
|
"""Pluggable auth modules for Home Assistant."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import importlib
|
import importlib
|
||||||
import logging
|
import logging
|
||||||
import types
|
import types
|
||||||
@ -66,7 +68,7 @@ class MultiFactorAuthModule:
|
|||||||
"""Return a voluptuous schema to define mfa auth module's input."""
|
"""Return a voluptuous schema to define mfa auth module's input."""
|
||||||
raise NotImplementedError
|
raise NotImplementedError
|
||||||
|
|
||||||
async def async_setup_flow(self, user_id: str) -> "SetupFlow":
|
async def async_setup_flow(self, user_id: str) -> SetupFlow:
|
||||||
"""Return a data entry flow handler for setup module.
|
"""Return a data entry flow handler for setup module.
|
||||||
|
|
||||||
Mfa module should extend SetupFlow
|
Mfa module should extend SetupFlow
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Auth providers for Home Assistant."""
|
"""Auth providers for Home Assistant."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import importlib
|
import importlib
|
||||||
import logging
|
import logging
|
||||||
import types
|
import types
|
||||||
@ -92,7 +94,7 @@ class AuthProvider:
|
|||||||
|
|
||||||
# Implement by extending class
|
# Implement by extending class
|
||||||
|
|
||||||
async def async_login_flow(self, context: Optional[Dict]) -> "LoginFlow":
|
async def async_login_flow(self, context: Optional[Dict]) -> LoginFlow:
|
||||||
"""Return the data flow for logging in with auth provider.
|
"""Return the data flow for logging in with auth provider.
|
||||||
|
|
||||||
Auth provider should extend LoginFlow and return an instance.
|
Auth provider should extend LoginFlow and return an instance.
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Home Assistant auth provider."""
|
"""Home Assistant auth provider."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import base64
|
import base64
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
@ -31,7 +33,7 @@ CONFIG_SCHEMA = vol.All(AUTH_PROVIDER_SCHEMA, _disallow_id)
|
|||||||
|
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_get_provider(hass: HomeAssistant) -> "HassAuthProvider":
|
def async_get_provider(hass: HomeAssistant) -> HassAuthProvider:
|
||||||
"""Get the provider."""
|
"""Get the provider."""
|
||||||
for prv in hass.auth.auth_providers:
|
for prv in hass.auth.auth_providers:
|
||||||
if prv.type == "homeassistant":
|
if prv.type == "homeassistant":
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Manage config entries in Home Assistant."""
|
"""Manage config entries in Home Assistant."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import functools
|
import functools
|
||||||
import logging
|
import logging
|
||||||
@ -526,7 +528,7 @@ class ConfigEntriesFlowManager(data_entry_flow.FlowManager):
|
|||||||
|
|
||||||
async def async_create_flow(
|
async def async_create_flow(
|
||||||
self, handler_key: Any, *, context: Optional[Dict] = None, data: Any = None
|
self, handler_key: Any, *, context: Optional[Dict] = None, data: Any = None
|
||||||
) -> "ConfigFlow":
|
) -> ConfigFlow:
|
||||||
"""Create a flow for specified handler.
|
"""Create a flow for specified handler.
|
||||||
|
|
||||||
Handler key is the domain of the component that we want to set up.
|
Handler key is the domain of the component that we want to set up.
|
||||||
@ -890,7 +892,7 @@ class ConfigFlow(data_entry_flow.FlowHandler):
|
|||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
@callback
|
@callback
|
||||||
def async_get_options_flow(config_entry: ConfigEntry) -> "OptionsFlow":
|
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
|
||||||
"""Get the options flow for this handler."""
|
"""Get the options flow for this handler."""
|
||||||
raise data_entry_flow.UnknownHandler
|
raise data_entry_flow.UnknownHandler
|
||||||
|
|
||||||
@ -1074,7 +1076,7 @@ class OptionsFlowManager(data_entry_flow.FlowManager):
|
|||||||
*,
|
*,
|
||||||
context: Optional[Dict[str, Any]] = None,
|
context: Optional[Dict[str, Any]] = None,
|
||||||
data: Optional[Dict[str, Any]] = None,
|
data: Optional[Dict[str, Any]] = None,
|
||||||
) -> "OptionsFlow":
|
) -> OptionsFlow:
|
||||||
"""Create an options flow for a config entry.
|
"""Create an options flow for a config entry.
|
||||||
|
|
||||||
Entry_id and flow.handler is the same thing to map entry with flow.
|
Entry_id and flow.handler is the same thing to map entry with flow.
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Classes to help gather user submissions."""
|
"""Classes to help gather user submissions."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import abc
|
import abc
|
||||||
import asyncio
|
import asyncio
|
||||||
from typing import Any, Dict, List, Optional, cast
|
from typing import Any, Dict, List, Optional, cast
|
||||||
@ -75,7 +77,7 @@ class FlowManager(abc.ABC):
|
|||||||
*,
|
*,
|
||||||
context: Optional[Dict[str, Any]] = None,
|
context: Optional[Dict[str, Any]] = None,
|
||||||
data: Optional[Dict[str, Any]] = None,
|
data: Optional[Dict[str, Any]] = None,
|
||||||
) -> "FlowHandler":
|
) -> FlowHandler:
|
||||||
"""Create a flow for specified handler.
|
"""Create a flow for specified handler.
|
||||||
|
|
||||||
Handler key is the domain of the component that we want to set up.
|
Handler key is the domain of the component that we want to set up.
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Helper to check the configuration file."""
|
"""Helper to check the configuration file."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
@ -49,7 +51,7 @@ class HomeAssistantConfig(OrderedDict):
|
|||||||
message: str,
|
message: str,
|
||||||
domain: Optional[str] = None,
|
domain: Optional[str] = None,
|
||||||
config: Optional[ConfigType] = None,
|
config: Optional[ConfigType] = None,
|
||||||
) -> "HomeAssistantConfig":
|
) -> HomeAssistantConfig:
|
||||||
"""Add a single error."""
|
"""Add a single error."""
|
||||||
self.errors.append(CheckConfigError(str(message), domain, config))
|
self.errors.append(CheckConfigError(str(message), domain, config))
|
||||||
return self
|
return self
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Class to manage the entities for a single platform."""
|
"""Class to manage the entities for a single platform."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
from contextvars import ContextVar
|
from contextvars import ContextVar
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
@ -245,7 +247,7 @@ class EntityPlatform:
|
|||||||
warn_task.cancel()
|
warn_task.cancel()
|
||||||
|
|
||||||
def _schedule_add_entities(
|
def _schedule_add_entities(
|
||||||
self, new_entities: Iterable["Entity"], update_before_add: bool = False
|
self, new_entities: Iterable[Entity], update_before_add: bool = False
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Schedule adding entities for a single platform, synchronously."""
|
"""Schedule adding entities for a single platform, synchronously."""
|
||||||
run_callback_threadsafe(
|
run_callback_threadsafe(
|
||||||
@ -257,7 +259,7 @@ class EntityPlatform:
|
|||||||
|
|
||||||
@callback
|
@callback
|
||||||
def _async_schedule_add_entities(
|
def _async_schedule_add_entities(
|
||||||
self, new_entities: Iterable["Entity"], update_before_add: bool = False
|
self, new_entities: Iterable[Entity], update_before_add: bool = False
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Schedule adding entities for a single platform async."""
|
"""Schedule adding entities for a single platform async."""
|
||||||
task = self.hass.async_create_task(
|
task = self.hass.async_create_task(
|
||||||
@ -268,7 +270,7 @@ class EntityPlatform:
|
|||||||
self._tasks.append(task)
|
self._tasks.append(task)
|
||||||
|
|
||||||
def add_entities(
|
def add_entities(
|
||||||
self, new_entities: Iterable["Entity"], update_before_add: bool = False
|
self, new_entities: Iterable[Entity], update_before_add: bool = False
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Add entities for a single platform."""
|
"""Add entities for a single platform."""
|
||||||
# That avoid deadlocks
|
# That avoid deadlocks
|
||||||
@ -284,7 +286,7 @@ class EntityPlatform:
|
|||||||
).result()
|
).result()
|
||||||
|
|
||||||
async def async_add_entities(
|
async def async_add_entities(
|
||||||
self, new_entities: Iterable["Entity"], update_before_add: bool = False
|
self, new_entities: Iterable[Entity], update_before_add: bool = False
|
||||||
) -> None:
|
) -> None:
|
||||||
"""Add entities for a single platform async.
|
"""Add entities for a single platform async.
|
||||||
|
|
||||||
@ -547,7 +549,7 @@ class EntityPlatform:
|
|||||||
|
|
||||||
async def async_extract_from_service(
|
async def async_extract_from_service(
|
||||||
self, service_call: ServiceCall, expand_group: bool = True
|
self, service_call: ServiceCall, expand_group: bool = True
|
||||||
) -> List["Entity"]:
|
) -> List[Entity]:
|
||||||
"""Extract all known and available entities from a service call.
|
"""Extract all known and available entities from a service call.
|
||||||
|
|
||||||
Will return an empty list if entities specified but unknown.
|
Will return an empty list if entities specified but unknown.
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Module to coordinate user intentions."""
|
"""Module to coordinate user intentions."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import re
|
import re
|
||||||
from typing import Any, Callable, Dict, Iterable, Optional
|
from typing import Any, Callable, Dict, Iterable, Optional
|
||||||
@ -29,7 +31,7 @@ SPEECH_TYPE_SSML = "ssml"
|
|||||||
|
|
||||||
@callback
|
@callback
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def async_register(hass: HomeAssistantType, handler: "IntentHandler") -> None:
|
def async_register(hass: HomeAssistantType, handler: IntentHandler) -> None:
|
||||||
"""Register an intent with Home Assistant."""
|
"""Register an intent with Home Assistant."""
|
||||||
intents = hass.data.get(DATA_KEY)
|
intents = hass.data.get(DATA_KEY)
|
||||||
if intents is None:
|
if intents is None:
|
||||||
@ -53,7 +55,7 @@ async def async_handle(
|
|||||||
slots: Optional[_SlotsType] = None,
|
slots: Optional[_SlotsType] = None,
|
||||||
text_input: Optional[str] = None,
|
text_input: Optional[str] = None,
|
||||||
context: Optional[Context] = None,
|
context: Optional[Context] = None,
|
||||||
) -> "IntentResponse":
|
) -> IntentResponse:
|
||||||
"""Handle an intent."""
|
"""Handle an intent."""
|
||||||
handler: IntentHandler = hass.data.get(DATA_KEY, {}).get(intent_type)
|
handler: IntentHandler = hass.data.get(DATA_KEY, {}).get(intent_type)
|
||||||
|
|
||||||
@ -131,7 +133,7 @@ class IntentHandler:
|
|||||||
platforms: Optional[Iterable[str]] = []
|
platforms: Optional[Iterable[str]] = []
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def async_can_handle(self, intent_obj: "Intent") -> bool:
|
def async_can_handle(self, intent_obj: Intent) -> bool:
|
||||||
"""Test if an intent can be handled."""
|
"""Test if an intent can be handled."""
|
||||||
return self.platforms is None or intent_obj.platform in self.platforms
|
return self.platforms is None or intent_obj.platform in self.platforms
|
||||||
|
|
||||||
@ -152,7 +154,7 @@ class IntentHandler:
|
|||||||
|
|
||||||
return self._slot_schema(slots) # type: ignore
|
return self._slot_schema(slots) # type: ignore
|
||||||
|
|
||||||
async def async_handle(self, intent_obj: "Intent") -> "IntentResponse":
|
async def async_handle(self, intent_obj: Intent) -> IntentResponse:
|
||||||
"""Handle the intent."""
|
"""Handle the intent."""
|
||||||
raise NotImplementedError()
|
raise NotImplementedError()
|
||||||
|
|
||||||
@ -195,7 +197,7 @@ class ServiceIntentHandler(IntentHandler):
|
|||||||
self.service = service
|
self.service = service
|
||||||
self.speech = speech
|
self.speech = speech
|
||||||
|
|
||||||
async def async_handle(self, intent_obj: "Intent") -> "IntentResponse":
|
async def async_handle(self, intent_obj: Intent) -> IntentResponse:
|
||||||
"""Handle the hass intent."""
|
"""Handle the hass intent."""
|
||||||
hass = intent_obj.hass
|
hass = intent_obj.hass
|
||||||
slots = self.async_validate_slots(intent_obj.slots)
|
slots = self.async_validate_slots(intent_obj.slots)
|
||||||
@ -236,7 +238,7 @@ class Intent:
|
|||||||
self.context = context
|
self.context = context
|
||||||
|
|
||||||
@callback
|
@callback
|
||||||
def create_response(self) -> "IntentResponse":
|
def create_response(self) -> IntentResponse:
|
||||||
"""Create a response."""
|
"""Create a response."""
|
||||||
return IntentResponse(self)
|
return IntentResponse(self)
|
||||||
|
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Support for restoring entity states on startup."""
|
"""Support for restoring entity states on startup."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
from datetime import datetime, timedelta
|
from datetime import datetime, timedelta
|
||||||
import logging
|
import logging
|
||||||
@ -48,7 +50,7 @@ class StoredState:
|
|||||||
return {"state": self.state.as_dict(), "last_seen": self.last_seen}
|
return {"state": self.state.as_dict(), "last_seen": self.last_seen}
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_dict(cls, json_dict: Dict) -> "StoredState":
|
def from_dict(cls, json_dict: Dict) -> StoredState:
|
||||||
"""Initialize a stored state from a dict."""
|
"""Initialize a stored state from a dict."""
|
||||||
last_seen = json_dict["last_seen"]
|
last_seen = json_dict["last_seen"]
|
||||||
|
|
||||||
@ -62,11 +64,11 @@ class RestoreStateData:
|
|||||||
"""Helper class for managing the helper saved data."""
|
"""Helper class for managing the helper saved data."""
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
async def async_get_instance(cls, hass: HomeAssistant) -> "RestoreStateData":
|
async def async_get_instance(cls, hass: HomeAssistant) -> RestoreStateData:
|
||||||
"""Get the singleton instance of this data helper."""
|
"""Get the singleton instance of this data helper."""
|
||||||
|
|
||||||
@singleton(DATA_RESTORE_STATE_TASK)
|
@singleton(DATA_RESTORE_STATE_TASK)
|
||||||
async def load_instance(hass: HomeAssistant) -> "RestoreStateData":
|
async def load_instance(hass: HomeAssistant) -> RestoreStateData:
|
||||||
"""Get the singleton instance of this data helper."""
|
"""Get the singleton instance of this data helper."""
|
||||||
data = cls(hass)
|
data = cls(hass)
|
||||||
|
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Service calling related helpers."""
|
"""Service calling related helpers."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import dataclasses
|
import dataclasses
|
||||||
from functools import partial, wraps
|
from functools import partial, wraps
|
||||||
@ -230,10 +232,10 @@ def extract_entity_ids(
|
|||||||
@bind_hass
|
@bind_hass
|
||||||
async def async_extract_entities(
|
async def async_extract_entities(
|
||||||
hass: HomeAssistantType,
|
hass: HomeAssistantType,
|
||||||
entities: Iterable["Entity"],
|
entities: Iterable[Entity],
|
||||||
service_call: ha.ServiceCall,
|
service_call: ha.ServiceCall,
|
||||||
expand_group: bool = True,
|
expand_group: bool = True,
|
||||||
) -> List["Entity"]:
|
) -> List[Entity]:
|
||||||
"""Extract a list of entity objects from a service call.
|
"""Extract a list of entity objects from a service call.
|
||||||
|
|
||||||
Will convert group entity ids to the entity ids it represents.
|
Will convert group entity ids to the entity ids it represents.
|
||||||
@ -634,7 +636,7 @@ async def entity_service_call(
|
|||||||
|
|
||||||
async def _handle_entity_call(
|
async def _handle_entity_call(
|
||||||
hass: HomeAssistantType,
|
hass: HomeAssistantType,
|
||||||
entity: "Entity",
|
entity: Entity,
|
||||||
func: Union[str, Callable[..., Any]],
|
func: Union[str, Callable[..., Any]],
|
||||||
data: Union[Dict, ha.ServiceCall],
|
data: Union[Dict, ha.ServiceCall],
|
||||||
context: ha.Context,
|
context: ha.Context,
|
||||||
|
@ -26,6 +26,8 @@ The following cases will never be passed to your function:
|
|||||||
- if either state is unknown/unavailable
|
- if either state is unknown/unavailable
|
||||||
- state adding/removing
|
- state adding/removing
|
||||||
"""
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
from types import MappingProxyType
|
from types import MappingProxyType
|
||||||
from typing import Any, Callable, Dict, Optional, Tuple, Union
|
from typing import Any, Callable, Dict, Optional, Tuple, Union
|
||||||
|
|
||||||
@ -65,7 +67,7 @@ async def create_checker(
|
|||||||
hass: HomeAssistant,
|
hass: HomeAssistant,
|
||||||
_domain: str,
|
_domain: str,
|
||||||
extra_significant_check: Optional[ExtraCheckTypeFunc] = None,
|
extra_significant_check: Optional[ExtraCheckTypeFunc] = None,
|
||||||
) -> "SignificantlyChangedChecker":
|
) -> SignificantlyChangedChecker:
|
||||||
"""Create a significantly changed checker for a domain."""
|
"""Create a significantly changed checker for a domain."""
|
||||||
await _initialize(hass)
|
await _initialize(hass)
|
||||||
return SignificantlyChangedChecker(hass, extra_significant_check)
|
return SignificantlyChangedChecker(hass, extra_significant_check)
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Helpers for sun events."""
|
"""Helpers for sun events."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import datetime
|
import datetime
|
||||||
from typing import TYPE_CHECKING, Optional, Union
|
from typing import TYPE_CHECKING, Optional, Union
|
||||||
|
|
||||||
@ -17,7 +19,7 @@ DATA_LOCATION_CACHE = "astral_location_cache"
|
|||||||
|
|
||||||
@callback
|
@callback
|
||||||
@bind_hass
|
@bind_hass
|
||||||
def get_astral_location(hass: HomeAssistantType) -> "astral.Location":
|
def get_astral_location(hass: HomeAssistantType) -> astral.Location:
|
||||||
"""Get an astral location for the current Home Assistant configuration."""
|
"""Get an astral location for the current Home Assistant configuration."""
|
||||||
from astral import Location # pylint: disable=import-outside-toplevel
|
from astral import Location # pylint: disable=import-outside-toplevel
|
||||||
|
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Template helper methods for rendering strings with Home Assistant data."""
|
"""Template helper methods for rendering strings with Home Assistant data."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
from ast import literal_eval
|
from ast import literal_eval
|
||||||
import asyncio
|
import asyncio
|
||||||
import base64
|
import base64
|
||||||
@ -155,7 +157,7 @@ class TupleWrapper(tuple, ResultWrapper):
|
|||||||
|
|
||||||
def __new__(
|
def __new__(
|
||||||
cls, value: tuple, *, render_result: Optional[str] = None
|
cls, value: tuple, *, render_result: Optional[str] = None
|
||||||
) -> "TupleWrapper":
|
) -> TupleWrapper:
|
||||||
"""Create a new tuple class."""
|
"""Create a new tuple class."""
|
||||||
return super().__new__(cls, tuple(value))
|
return super().__new__(cls, tuple(value))
|
||||||
|
|
||||||
@ -297,7 +299,7 @@ class Template:
|
|||||||
self._limited = None
|
self._limited = None
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def _env(self) -> "TemplateEnvironment":
|
def _env(self) -> TemplateEnvironment:
|
||||||
if self.hass is None or self._limited:
|
if self.hass is None or self._limited:
|
||||||
return _NO_HASS_ENV
|
return _NO_HASS_ENV
|
||||||
ret: Optional[TemplateEnvironment] = self.hass.data.get(_ENVIRONMENT)
|
ret: Optional[TemplateEnvironment] = self.hass.data.get(_ENVIRONMENT)
|
||||||
@ -530,7 +532,7 @@ class Template:
|
|||||||
)
|
)
|
||||||
return value if error_value is _SENTINEL else error_value
|
return value if error_value is _SENTINEL else error_value
|
||||||
|
|
||||||
def _ensure_compiled(self, limited: bool = False) -> "Template":
|
def _ensure_compiled(self, limited: bool = False) -> Template:
|
||||||
"""Bind a template to a specific hass instance."""
|
"""Bind a template to a specific hass instance."""
|
||||||
self.ensure_valid()
|
self.ensure_valid()
|
||||||
|
|
||||||
|
@ -4,6 +4,8 @@ The methods for loading Home Assistant integrations.
|
|||||||
This module has quite some complex parts. I have tried to add as much
|
This module has quite some complex parts. I have tried to add as much
|
||||||
documentation as possible to keep it understandable.
|
documentation as possible to keep it understandable.
|
||||||
"""
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import functools as ft
|
import functools as ft
|
||||||
import importlib
|
import importlib
|
||||||
@ -114,7 +116,7 @@ def manifest_from_legacy_module(domain: str, module: ModuleType) -> Manifest:
|
|||||||
|
|
||||||
async def _async_get_custom_components(
|
async def _async_get_custom_components(
|
||||||
hass: "HomeAssistant",
|
hass: "HomeAssistant",
|
||||||
) -> Dict[str, "Integration"]:
|
) -> Dict[str, Integration]:
|
||||||
"""Return list of custom integrations."""
|
"""Return list of custom integrations."""
|
||||||
if hass.config.safe_mode:
|
if hass.config.safe_mode:
|
||||||
return {}
|
return {}
|
||||||
@ -155,7 +157,7 @@ async def _async_get_custom_components(
|
|||||||
|
|
||||||
async def async_get_custom_components(
|
async def async_get_custom_components(
|
||||||
hass: "HomeAssistant",
|
hass: "HomeAssistant",
|
||||||
) -> Dict[str, "Integration"]:
|
) -> Dict[str, Integration]:
|
||||||
"""Return cached list of custom integrations."""
|
"""Return cached list of custom integrations."""
|
||||||
reg_or_evt = hass.data.get(DATA_CUSTOM_COMPONENTS)
|
reg_or_evt = hass.data.get(DATA_CUSTOM_COMPONENTS)
|
||||||
|
|
||||||
@ -175,7 +177,7 @@ async def async_get_custom_components(
|
|||||||
return cast(Dict[str, "Integration"], reg_or_evt)
|
return cast(Dict[str, "Integration"], reg_or_evt)
|
||||||
|
|
||||||
|
|
||||||
async def async_get_config_flows(hass: "HomeAssistant") -> Set[str]:
|
async def async_get_config_flows(hass: HomeAssistant) -> Set[str]:
|
||||||
"""Return cached list of config flows."""
|
"""Return cached list of config flows."""
|
||||||
# pylint: disable=import-outside-toplevel
|
# pylint: disable=import-outside-toplevel
|
||||||
from homeassistant.generated.config_flows import FLOWS
|
from homeassistant.generated.config_flows import FLOWS
|
||||||
@ -195,7 +197,7 @@ async def async_get_config_flows(hass: "HomeAssistant") -> Set[str]:
|
|||||||
return flows
|
return flows
|
||||||
|
|
||||||
|
|
||||||
async def async_get_zeroconf(hass: "HomeAssistant") -> Dict[str, List[Dict[str, str]]]:
|
async def async_get_zeroconf(hass: HomeAssistant) -> Dict[str, List[Dict[str, str]]]:
|
||||||
"""Return cached list of zeroconf types."""
|
"""Return cached list of zeroconf types."""
|
||||||
zeroconf: Dict[str, List[Dict[str, str]]] = ZEROCONF.copy()
|
zeroconf: Dict[str, List[Dict[str, str]]] = ZEROCONF.copy()
|
||||||
|
|
||||||
@ -218,7 +220,7 @@ async def async_get_zeroconf(hass: "HomeAssistant") -> Dict[str, List[Dict[str,
|
|||||||
return zeroconf
|
return zeroconf
|
||||||
|
|
||||||
|
|
||||||
async def async_get_dhcp(hass: "HomeAssistant") -> List[Dict[str, str]]:
|
async def async_get_dhcp(hass: HomeAssistant) -> List[Dict[str, str]]:
|
||||||
"""Return cached list of dhcp types."""
|
"""Return cached list of dhcp types."""
|
||||||
dhcp: List[Dict[str, str]] = DHCP.copy()
|
dhcp: List[Dict[str, str]] = DHCP.copy()
|
||||||
|
|
||||||
@ -232,7 +234,7 @@ async def async_get_dhcp(hass: "HomeAssistant") -> List[Dict[str, str]]:
|
|||||||
return dhcp
|
return dhcp
|
||||||
|
|
||||||
|
|
||||||
async def async_get_homekit(hass: "HomeAssistant") -> Dict[str, str]:
|
async def async_get_homekit(hass: HomeAssistant) -> Dict[str, str]:
|
||||||
"""Return cached list of homekit models."""
|
"""Return cached list of homekit models."""
|
||||||
|
|
||||||
homekit: Dict[str, str] = HOMEKIT.copy()
|
homekit: Dict[str, str] = HOMEKIT.copy()
|
||||||
@ -251,7 +253,7 @@ async def async_get_homekit(hass: "HomeAssistant") -> Dict[str, str]:
|
|||||||
return homekit
|
return homekit
|
||||||
|
|
||||||
|
|
||||||
async def async_get_ssdp(hass: "HomeAssistant") -> Dict[str, List[Dict[str, str]]]:
|
async def async_get_ssdp(hass: HomeAssistant) -> Dict[str, List[Dict[str, str]]]:
|
||||||
"""Return cached list of ssdp mappings."""
|
"""Return cached list of ssdp mappings."""
|
||||||
|
|
||||||
ssdp: Dict[str, List[Dict[str, str]]] = SSDP.copy()
|
ssdp: Dict[str, List[Dict[str, str]]] = SSDP.copy()
|
||||||
@ -266,7 +268,7 @@ async def async_get_ssdp(hass: "HomeAssistant") -> Dict[str, List[Dict[str, str]
|
|||||||
return ssdp
|
return ssdp
|
||||||
|
|
||||||
|
|
||||||
async def async_get_mqtt(hass: "HomeAssistant") -> Dict[str, List[str]]:
|
async def async_get_mqtt(hass: HomeAssistant) -> Dict[str, List[str]]:
|
||||||
"""Return cached list of MQTT mappings."""
|
"""Return cached list of MQTT mappings."""
|
||||||
|
|
||||||
mqtt: Dict[str, List[str]] = MQTT.copy()
|
mqtt: Dict[str, List[str]] = MQTT.copy()
|
||||||
@ -287,7 +289,7 @@ class Integration:
|
|||||||
@classmethod
|
@classmethod
|
||||||
def resolve_from_root(
|
def resolve_from_root(
|
||||||
cls, hass: "HomeAssistant", root_module: ModuleType, domain: str
|
cls, hass: "HomeAssistant", root_module: ModuleType, domain: str
|
||||||
) -> "Optional[Integration]":
|
) -> Optional[Integration]:
|
||||||
"""Resolve an integration from a root module."""
|
"""Resolve an integration from a root module."""
|
||||||
for base in root_module.__path__: # type: ignore
|
for base in root_module.__path__: # type: ignore
|
||||||
manifest_path = pathlib.Path(base) / domain / "manifest.json"
|
manifest_path = pathlib.Path(base) / domain / "manifest.json"
|
||||||
@ -312,7 +314,7 @@ class Integration:
|
|||||||
@classmethod
|
@classmethod
|
||||||
def resolve_legacy(
|
def resolve_legacy(
|
||||||
cls, hass: "HomeAssistant", domain: str
|
cls, hass: "HomeAssistant", domain: str
|
||||||
) -> "Optional[Integration]":
|
) -> Optional[Integration]:
|
||||||
"""Resolve legacy component.
|
"""Resolve legacy component.
|
||||||
|
|
||||||
Will create a stub manifest.
|
Will create a stub manifest.
|
||||||
@ -671,7 +673,7 @@ class ModuleWrapper:
|
|||||||
class Components:
|
class Components:
|
||||||
"""Helper to load components."""
|
"""Helper to load components."""
|
||||||
|
|
||||||
def __init__(self, hass: "HomeAssistant") -> None:
|
def __init__(self, hass: HomeAssistant) -> None:
|
||||||
"""Initialize the Components class."""
|
"""Initialize the Components class."""
|
||||||
self._hass = hass
|
self._hass = hass
|
||||||
|
|
||||||
@ -697,7 +699,7 @@ class Components:
|
|||||||
class Helpers:
|
class Helpers:
|
||||||
"""Helper to load helpers."""
|
"""Helper to load helpers."""
|
||||||
|
|
||||||
def __init__(self, hass: "HomeAssistant") -> None:
|
def __init__(self, hass: HomeAssistant) -> None:
|
||||||
"""Initialize the Helpers class."""
|
"""Initialize the Helpers class."""
|
||||||
self._hass = hass
|
self._hass = hass
|
||||||
|
|
||||||
@ -758,7 +760,7 @@ async def _async_component_dependencies(
|
|||||||
return loaded
|
return loaded
|
||||||
|
|
||||||
|
|
||||||
def _async_mount_config_dir(hass: "HomeAssistant") -> bool:
|
def _async_mount_config_dir(hass: HomeAssistant) -> bool:
|
||||||
"""Mount config dir in order to load custom_component.
|
"""Mount config dir in order to load custom_component.
|
||||||
|
|
||||||
Async friendly but not a coroutine.
|
Async friendly but not a coroutine.
|
||||||
@ -771,7 +773,7 @@ def _async_mount_config_dir(hass: "HomeAssistant") -> bool:
|
|||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
def _lookup_path(hass: "HomeAssistant") -> List[str]:
|
def _lookup_path(hass: HomeAssistant) -> List[str]:
|
||||||
"""Return the lookup paths for legacy lookups."""
|
"""Return the lookup paths for legacy lookups."""
|
||||||
if hass.config.safe_mode:
|
if hass.config.safe_mode:
|
||||||
return [PACKAGE_BUILTIN]
|
return [PACKAGE_BUILTIN]
|
||||||
|
@ -48,7 +48,7 @@ class MockRequest:
|
|||||||
self.mock_source = mock_source
|
self.mock_source = mock_source
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def query(self) -> "MultiDict[str]":
|
def query(self) -> MultiDict[str]:
|
||||||
"""Return a dictionary with the query variables."""
|
"""Return a dictionary with the query variables."""
|
||||||
return MultiDict(parse_qsl(self.query_string, keep_blank_values=True))
|
return MultiDict(parse_qsl(self.query_string, keep_blank_values=True))
|
||||||
|
|
||||||
@ -66,7 +66,7 @@ class MockRequest:
|
|||||||
"""Return the body as JSON."""
|
"""Return the body as JSON."""
|
||||||
return json.loads(self._text)
|
return json.loads(self._text)
|
||||||
|
|
||||||
async def post(self) -> "MultiDict[str]":
|
async def post(self) -> MultiDict[str]:
|
||||||
"""Return POST parameters."""
|
"""Return POST parameters."""
|
||||||
return MultiDict(parse_qsl(self._text, keep_blank_values=True))
|
return MultiDict(parse_qsl(self._text, keep_blank_values=True))
|
||||||
|
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""Custom yaml object types."""
|
"""Custom yaml object types."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
import yaml
|
import yaml
|
||||||
@ -19,6 +21,6 @@ class Input:
|
|||||||
name: str
|
name: str
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def from_node(cls, loader: yaml.Loader, node: yaml.nodes.Node) -> "Input":
|
def from_node(cls, loader: yaml.Loader, node: yaml.nodes.Node) -> Input:
|
||||||
"""Create a new placeholder from a node."""
|
"""Create a new placeholder from a node."""
|
||||||
return cls(node.value)
|
return cls(node.value)
|
||||||
|
@ -1,4 +1,6 @@
|
|||||||
"""API for Lokalise."""
|
"""API for Lokalise."""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
from pprint import pprint
|
from pprint import pprint
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
@ -6,7 +8,7 @@ import requests
|
|||||||
from .util import get_lokalise_token
|
from .util import get_lokalise_token
|
||||||
|
|
||||||
|
|
||||||
def get_api(project_id, debug=False) -> "Lokalise":
|
def get_api(project_id, debug=False) -> Lokalise:
|
||||||
"""Get Lokalise API."""
|
"""Get Lokalise API."""
|
||||||
return Lokalise(project_id, get_lokalise_token(), debug)
|
return Lokalise(project_id, get_lokalise_token(), debug)
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user