Use dataclass for MqttServiceInfo (#60191)

* Use dataclass for MqttServiceInfo

* Drop test exception

Co-authored-by: epenet <epenet@users.noreply.github.com>
This commit is contained in:
epenet 2021-11-23 09:01:40 +01:00 committed by GitHub
parent 42ed6ddba3
commit 3b0d984959
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -1,13 +1,14 @@
"""Support for MQTT discovery.""" """Support for MQTT discovery."""
import asyncio import asyncio
from collections import deque from collections import deque
from dataclasses import dataclass
import datetime as dt import datetime as dt
import functools import functools
import json import json
import logging import logging
import re import re
import time import time
from typing import TypedDict from typing import Any
from homeassistant.const import CONF_DEVICE, CONF_PLATFORM from homeassistant.const import CONF_DEVICE, CONF_PLATFORM
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
@ -17,6 +18,7 @@ from homeassistant.helpers.dispatcher import (
async_dispatcher_connect, async_dispatcher_connect,
async_dispatcher_send, async_dispatcher_send,
) )
from homeassistant.helpers.frame import report
from homeassistant.loader import async_get_mqtt from homeassistant.loader import async_get_mqtt
from .. import mqtt from .. import mqtt
@ -89,7 +91,8 @@ class MQTTConfig(dict):
"""Dummy class to allow adding attributes.""" """Dummy class to allow adding attributes."""
class MqttServiceInfo(TypedDict): @dataclass
class MqttServiceInfo:
"""Prepared info from mqtt entries.""" """Prepared info from mqtt entries."""
topic: str topic: str
@ -99,6 +102,24 @@ class MqttServiceInfo(TypedDict):
subscribed_topic: str subscribed_topic: str
timestamp: dt.datetime timestamp: dt.datetime
# Used to prevent log flooding. To be removed in 2022.6
_warning_logged: bool = False
def __getitem__(self, name: str) -> Any:
"""
Allow property access by name for compatibility reason.
Deprecated, and will be removed in version 2022.6.
"""
if not self._warning_logged:
report(
f"accessed discovery_info['{name}'] instead of discovery_info.{name}; this will fail in version 2022.6",
exclude_integrations={"mqtt"},
error_if_core=False,
)
self._warning_logged = True
return getattr(self, name)
async def async_start( # noqa: C901 async def async_start( # noqa: C901
hass: HomeAssistant, discovery_topic, config_entry=None hass: HomeAssistant, discovery_topic, config_entry=None