Add TYPE_CHECKING condition on type assertions for mqtt (#100107)

Add TYPE_CHECKING condition on type assertions
This commit is contained in:
Jan Bouwhuis 2023-09-11 10:58:33 +02:00 committed by GitHub
parent eb0099dee8
commit 20d0ebe3fa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 39 additions and 26 deletions

View File

@ -5,7 +5,7 @@ import asyncio
from collections.abc import Callable
from datetime import datetime
import logging
from typing import Any, TypeVar, cast
from typing import TYPE_CHECKING, Any, TypeVar, cast
import jinja2
import voluptuous as vol
@ -313,7 +313,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
)
return
assert msg_topic is not None
if TYPE_CHECKING:
assert msg_topic is not None
await mqtt_data.client.async_publish(msg_topic, payload, qos, retain)
hass.services.async_register(

View File

@ -4,6 +4,7 @@ from __future__ import annotations
from base64 import b64decode
import functools
import logging
from typing import TYPE_CHECKING
import voluptuous as vol
@ -112,7 +113,8 @@ class MqttCamera(MqttEntity, Camera):
if CONF_IMAGE_ENCODING in self._config:
self._last_image = b64decode(msg.payload)
else:
assert isinstance(msg.payload, bytes)
if TYPE_CHECKING:
assert isinstance(msg.payload, bytes)
self._last_image = msg.payload
self._sub_state = subscription.async_prepare_subscribe_topics(

View File

@ -6,7 +6,7 @@ from collections.abc import Callable
import queue
from ssl import PROTOCOL_TLS_CLIENT, SSLContext, SSLError
from types import MappingProxyType
from typing import Any
from typing import TYPE_CHECKING, Any
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from cryptography.x509 import load_pem_x509_certificate
@ -224,7 +224,8 @@ class FlowHandler(ConfigFlow, domain=DOMAIN):
) -> FlowResult:
"""Confirm a Hass.io discovery."""
errors: dict[str, str] = {}
assert self._hassio_discovery
if TYPE_CHECKING:
assert self._hassio_discovery
if user_input is not None:
data: dict[str, Any] = self._hassio_discovery.copy()
@ -312,7 +313,8 @@ class MQTTOptionsFlowHandler(OptionsFlow):
def _birth_will(birt_or_will: str) -> dict[str, Any]:
"""Return the user input for birth or will."""
assert user_input
if TYPE_CHECKING:
assert user_input
return {
ATTR_TOPIC: user_input[f"{birt_or_will}_topic"],
ATTR_PAYLOAD: user_input.get(f"{birt_or_will}_payload", ""),

View File

@ -5,7 +5,7 @@ from collections import deque
from collections.abc import Callable
import datetime as dt
from functools import wraps
from typing import Any
from typing import TYPE_CHECKING, Any
import attr
@ -128,11 +128,11 @@ def update_entity_discovery_data(
hass: HomeAssistant, discovery_payload: DiscoveryInfoType, entity_id: str
) -> None:
"""Update discovery data."""
assert (
discovery_data := get_mqtt_data(hass).debug_info_entities[entity_id][
"discovery_data"
]
) is not None
discovery_data = get_mqtt_data(hass).debug_info_entities[entity_id][
"discovery_data"
]
if TYPE_CHECKING:
assert discovery_data is not None
discovery_data[ATTR_DISCOVERY_PAYLOAD] = discovery_payload

View File

@ -3,7 +3,7 @@ from __future__ import annotations
from collections.abc import Callable
import logging
from typing import Any
from typing import TYPE_CHECKING, Any
import attr
import voluptuous as vol
@ -269,7 +269,8 @@ async def async_setup_trigger(
config = TRIGGER_DISCOVERY_SCHEMA(config)
device_id = update_device(hass, config_entry, config)
assert isinstance(device_id, str)
if TYPE_CHECKING:
assert isinstance(device_id, str)
mqtt_device_trigger = MqttDeviceTrigger(
hass, config, device_id, discovery_data, config_entry
)
@ -286,7 +287,8 @@ async def async_removed_from_device(hass: HomeAssistant, device_id: str) -> None
if device_trigger:
device_trigger.detach_trigger()
discovery_data = device_trigger.discovery_data
assert discovery_data is not None
if TYPE_CHECKING:
assert discovery_data is not None
discovery_hash = discovery_data[ATTR_DISCOVERY_HASH]
debug_info.remove_trigger_discovery_data(hass, discovery_hash)

View File

@ -1,7 +1,7 @@
"""Diagnostics support for MQTT."""
from __future__ import annotations
from typing import Any
from typing import TYPE_CHECKING, Any
from homeassistant.components import device_tracker
from homeassistant.components.diagnostics import async_redact_data
@ -45,7 +45,8 @@ def _async_get_diagnostics(
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
mqtt_instance = get_mqtt_data(hass).client
assert mqtt_instance is not None
if TYPE_CHECKING:
assert mqtt_instance is not None
redacted_config = async_redact_data(mqtt_instance.conf, REDACT_CONFIG)

View File

@ -7,7 +7,7 @@ import functools
import logging
import re
import time
from typing import Any
from typing import TYPE_CHECKING, Any
import voluptuous as vol
@ -343,7 +343,8 @@ async def async_start( # noqa: C901
integration: str, msg: ReceiveMessage
) -> None:
"""Process the received message."""
assert mqtt_data.data_config_flow_lock
if TYPE_CHECKING:
assert mqtt_data.data_config_flow_lock
key = f"{integration}_{msg.subscribed_topic}"
# Lock to prevent initiating many parallel config flows.

View File

@ -6,7 +6,7 @@ import binascii
from collections.abc import Callable
import functools
import logging
from typing import Any
from typing import TYPE_CHECKING, Any
import httpx
import voluptuous as vol
@ -172,7 +172,8 @@ class MqttImage(MqttEntity, ImageEntity):
if CONF_IMAGE_ENCODING in self._config:
self._last_image = b64decode(msg.payload)
else:
assert isinstance(msg.payload, bytes)
if TYPE_CHECKING:
assert isinstance(msg.payload, bytes)
self._last_image = msg.payload
except (binascii.Error, ValueError, AssertionError) as err:
_LOGGER.error(

View File

@ -6,7 +6,7 @@ import asyncio
from collections.abc import Callable, Coroutine
from functools import partial
import logging
from typing import Any, Protocol, cast, final
from typing import TYPE_CHECKING, Any, Protocol, cast, final
import voluptuous as vol
@ -850,7 +850,8 @@ class MqttDiscoveryUpdate(Entity):
discovery_hash,
payload,
)
assert self._discovery_data
if TYPE_CHECKING:
assert self._discovery_data
old_payload: DiscoveryInfoType
old_payload = self._discovery_data[ATTR_DISCOVERY_PAYLOAD]
debug_info.update_entity_discovery_data(self.hass, payload, self.entity_id)
@ -877,7 +878,8 @@ class MqttDiscoveryUpdate(Entity):
send_discovery_done(self.hass, self._discovery_data)
if discovery_hash:
assert self._discovery_data is not None
if TYPE_CHECKING:
assert self._discovery_data is not None
debug_info.add_entity_discovery_data(
self.hass, self._discovery_data, self.entity_id
)

View File

@ -2,7 +2,7 @@
from __future__ import annotations
from collections.abc import Callable, Coroutine
from typing import Any
from typing import TYPE_CHECKING, Any
import attr
@ -31,7 +31,8 @@ class EntitySubscription:
) -> None:
"""Re-subscribe to the new topic if necessary."""
if not self._should_resubscribe(other):
assert other
if TYPE_CHECKING:
assert other
self.unsubscribe_callback = other.unsubscribe_callback
return