jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-28 03:07:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add pylint plugin to check if coordinator is placed in its own module (#108174)

Browse Source 
* Add pylint plugin to check if coordinator is placed in its own module

* Remove unintended changes

* Remove pylint disable and let CI only fail on W,E,F

* Make check conventional

* Apply review suggestion

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>

* Use option instead

* Remove pylint arguments from pre-commit

* Partially revert "Remove pylint disable and let CI only fail on W,E,F"

---------

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Jan-Philipp Benecke 2024-01-29 10:30:19 +01:00 committed by GitHub
parent 57622acabf
commit 95aea1488d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
62 changed files with 290 additions and 81 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.github/workflows/ci.yaml vendored
View File

@ -597,14 +597,14 @@ jobs:
run: | run: |
. venv/bin/activate . venv/bin/activate
python --version python --version
pylint --ignore-missing-annotations=y homeassistant pylint --ignore-missing-annotations=y --ignore-wrong-coordinator-module=y homeassistant
- name: Run pylint (partially) - name: Run pylint (partially)
if: needs.info.outputs.test_full_suite == 'false' if: needs.info.outputs.test_full_suite == 'false'
shell: bash shell: bash
run: | run: |
. venv/bin/activate . venv/bin/activate
python --version python --version
pylint --ignore-missing-annotations=y homeassistant/components/${{ needs.info.outputs.integrations_glob }} pylint --ignore-missing-annotations=y --ignore-wrong-coordinator-module=y homeassistant/components/${{ needs.info.outputs.integrations_glob }}
mypy: mypy:
name: Check mypy name: Check mypy

2
homeassistant/components/accuweather/__init__.py
View File

@ -75,7 +75,7 @@ async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
await hass.config_entries.async_reload(entry.entry_id) await hass.config_entries.async_reload(entry.entry_id)
class AccuWeatherDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): class AccuWeatherDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching AccuWeather data API.""" """Class to manage fetching AccuWeather data API."""
def __init__( def __init__(

2
homeassistant/components/aurora_abb_powerone/__init__.py
View File

@ -52,7 +52,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class AuroraAbbDataUpdateCoordinator(DataUpdateCoordinator[dict[str, float]]): class AuroraAbbDataUpdateCoordinator(DataUpdateCoordinator[dict[str, float]]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching AuroraAbbPowerone data.""" """Class to manage fetching AuroraAbbPowerone data."""
def __init__(self, hass: HomeAssistant, comport: str, address: int) -> None: def __init__(self, hass: HomeAssistant, comport: str, address: int) -> None:

2
homeassistant/components/brother/__init__.py
View File

@ -62,7 +62,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class BrotherDataUpdateCoordinator(DataUpdateCoordinator[BrotherSensors]): class BrotherDataUpdateCoordinator(DataUpdateCoordinator[BrotherSensors]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Brother data from the printer.""" """Class to manage fetching Brother data from the printer."""
def __init__(self, hass: HomeAssistant, brother: Brother) -> None: def __init__(self, hass: HomeAssistant, brother: Brother) -> None:

2
homeassistant/components/elmax/common.py
View File

@ -34,7 +34,7 @@ from .const import DEFAULT_TIMEOUT, DOMAIN
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]): class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Coordinator helper to handle Elmax API polling.""" """Coordinator helper to handle Elmax API polling."""
def __init__( def __init__(

2
homeassistant/components/environment_canada/__init__.py
View File

@ -99,7 +99,7 @@ def device_info(config_entry: ConfigEntry) -> DeviceInfo:
) )
class ECDataUpdateCoordinator(DataUpdateCoordinator): class ECDataUpdateCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching EC data.""" """Class to manage fetching EC data."""
def __init__(self, hass, ec_data, name, update_interval): def __init__(self, hass, ec_data, name, update_interval):

2
homeassistant/components/esphome/dashboard.py
View File

@ -158,7 +158,7 @@ async def async_set_dashboard_info(
await manager.async_set_dashboard_info(addon_slug, host, port) await manager.async_set_dashboard_info(addon_slug, host, port)
class ESPHomeDashboard(DataUpdateCoordinator[dict[str, ConfiguredDevice]]): class ESPHomeDashboard(DataUpdateCoordinator[dict[str, ConfiguredDevice]]): # pylint: disable=hass-enforce-coordinator-module
"""Class to interact with the ESPHome dashboard.""" """Class to interact with the ESPHome dashboard."""
def __init__( def __init__(

2
homeassistant/components/evil_genius_labs/__init__.py
View File

@ -50,7 +50,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class EvilGeniusUpdateCoordinator(DataUpdateCoordinator[dict]): class EvilGeniusUpdateCoordinator(DataUpdateCoordinator[dict]): # pylint: disable=hass-enforce-coordinator-module
"""Update coordinator for Evil Genius data.""" """Update coordinator for Evil Genius data."""
info: dict info: dict

2
homeassistant/components/flo/device.py
View File

@ -15,7 +15,7 @@ import homeassistant.util.dt as dt_util
from .const import DOMAIN as FLO_DOMAIN, LOGGER from .const import DOMAIN as FLO_DOMAIN, LOGGER
class FloDeviceDataUpdateCoordinator(DataUpdateCoordinator): class FloDeviceDataUpdateCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Flo device object.""" """Flo device object."""
def __init__( def __init__(

4
homeassistant/components/fritz/common.py
View File

@ -179,7 +179,7 @@ class UpdateCoordinatorDataType(TypedDict):
class FritzBoxTools( class FritzBoxTools(
update_coordinator.DataUpdateCoordinator[UpdateCoordinatorDataType] update_coordinator.DataUpdateCoordinator[UpdateCoordinatorDataType]
): ): # pylint: disable=hass-enforce-coordinator-module
"""FritzBoxTools class.""" """FritzBoxTools class."""
def __init__( def __init__(
@ -757,7 +757,7 @@ class FritzBoxTools(
raise HomeAssistantError("Service not supported") from ex raise HomeAssistantError("Service not supported") from ex
class AvmWrapper(FritzBoxTools): class AvmWrapper(FritzBoxTools): # pylint: disable=hass-enforce-coordinator-module
"""Setup AVM wrapper for API calls.""" """Setup AVM wrapper for API calls."""
async def _async_service_call( async def _async_service_call(

2
homeassistant/components/gios/__init__.py
View File

@ -74,7 +74,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class GiosDataUpdateCoordinator(DataUpdateCoordinator[GiosSensors]): class GiosDataUpdateCoordinator(DataUpdateCoordinator[GiosSensors]): # pylint: disable=hass-enforce-coordinator-module
"""Define an object to hold GIOS data.""" """Define an object to hold GIOS data."""
def __init__( def __init__(

2
homeassistant/components/gogogate2/common.py
View File

@ -47,7 +47,7 @@ class StateData(NamedTuple):
class DeviceDataUpdateCoordinator( class DeviceDataUpdateCoordinator(
DataUpdateCoordinator[GogoGate2InfoResponse | ISmartGateInfoResponse] DataUpdateCoordinator[GogoGate2InfoResponse | ISmartGateInfoResponse]
): ): # pylint: disable=hass-enforce-coordinator-module
"""Manages polling for state changes from the device.""" """Manages polling for state changes from the device."""
def __init__( def __init__(

4
homeassistant/components/google/calendar.py
View File

@ -265,7 +265,7 @@ def _truncate_timeline(timeline: Timeline, max_events: int) -> Timeline:
) )
class CalendarSyncUpdateCoordinator(DataUpdateCoordinator[Timeline]): class CalendarSyncUpdateCoordinator(DataUpdateCoordinator[Timeline]): # pylint: disable=hass-enforce-coordinator-module
"""Coordinator for calendar RPC calls that use an efficient sync.""" """Coordinator for calendar RPC calls that use an efficient sync."""
config_entry: ConfigEntry config_entry: ConfigEntry
@ -320,7 +320,7 @@ class CalendarSyncUpdateCoordinator(DataUpdateCoordinator[Timeline]):
return None return None
class CalendarQueryUpdateCoordinator(DataUpdateCoordinator[list[Event]]): class CalendarQueryUpdateCoordinator(DataUpdateCoordinator[list[Event]]): # pylint: disable=hass-enforce-coordinator-module
"""Coordinator for calendar RPC calls. """Coordinator for calendar RPC calls.
This sends a polling RPC, not using sync, as a workaround This sends a polling RPC, not using sync, as a workaround

2
homeassistant/components/gree/bridge.py
View File

@ -23,7 +23,7 @@ from .const import (
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class DeviceDataUpdateCoordinator(DataUpdateCoordinator): class DeviceDataUpdateCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Manages polling for state changes from the device.""" """Manages polling for state changes from the device."""
def __init__(self, hass: HomeAssistant, device: Device) -> None: def __init__(self, hass: HomeAssistant, device: Device) -> None:

2
homeassistant/components/hassio/__init__.py
View File

@ -746,7 +746,7 @@ def async_remove_addons_from_dev_reg(
dev_reg.async_remove_device(dev.id) dev_reg.async_remove_device(dev.id)
class HassioDataUpdateCoordinator(DataUpdateCoordinator): class HassioDataUpdateCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Class to retrieve Hass.io status.""" """Class to retrieve Hass.io status."""
def __init__( def __init__(

2
homeassistant/components/homeassistant_alerts/__init__.py
View File

@ -124,7 +124,7 @@ class IntegrationAlert:
return f"{self.filename}_{self.integration}" return f"{self.filename}_{self.integration}"
class AlertUpdateCoordinator(DataUpdateCoordinator[dict[str, IntegrationAlert]]): class AlertUpdateCoordinator(DataUpdateCoordinator[dict[str, IntegrationAlert]]): # pylint: disable=hass-enforce-coordinator-module
"""Data fetcher for HA Alerts.""" """Data fetcher for HA Alerts."""
def __init__(self, hass: HomeAssistant) -> None: def __init__(self, hass: HomeAssistant) -> None:

2
homeassistant/components/ialarm/__init__.py
View File

@ -53,7 +53,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class IAlarmDataUpdateCoordinator(DataUpdateCoordinator[None]): class IAlarmDataUpdateCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching iAlarm data.""" """Class to manage fetching iAlarm data."""
def __init__(self, hass: HomeAssistant, ialarm: IAlarm, mac: str) -> None: def __init__(self, hass: HomeAssistant, ialarm: IAlarm, mac: str) -> None:

2
homeassistant/components/idasen_desk/__init__.py
View File

@ -29,7 +29,7 @@ PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.COVER, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class IdasenDeskCoordinator(DataUpdateCoordinator[int | None]): class IdasenDeskCoordinator(DataUpdateCoordinator[int | None]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage updates for the Idasen Desk.""" """Class to manage updates for the Idasen Desk."""
def __init__( def __init__(

10
homeassistant/components/kostal_plenticore/helper.py
View File

@ -158,7 +158,7 @@ class DataUpdateCoordinatorMixin:
return True return True
class PlenticoreUpdateCoordinator(DataUpdateCoordinator[_DataT]): class PlenticoreUpdateCoordinator(DataUpdateCoordinator[_DataT]): # pylint: disable=hass-enforce-coordinator-module
"""Base implementation of DataUpdateCoordinator for Plenticore data.""" """Base implementation of DataUpdateCoordinator for Plenticore data."""
def __init__( def __init__(
@ -198,7 +198,7 @@ class PlenticoreUpdateCoordinator(DataUpdateCoordinator[_DataT]):
class ProcessDataUpdateCoordinator( class ProcessDataUpdateCoordinator(
PlenticoreUpdateCoordinator[Mapping[str, Mapping[str, str]]] PlenticoreUpdateCoordinator[Mapping[str, Mapping[str, str]]]
): ): # pylint: disable=hass-enforce-coordinator-module
"""Implementation of PlenticoreUpdateCoordinator for process data.""" """Implementation of PlenticoreUpdateCoordinator for process data."""
async def _async_update_data(self) -> dict[str, dict[str, str]]: async def _async_update_data(self) -> dict[str, dict[str, str]]:
@ -222,7 +222,7 @@ class ProcessDataUpdateCoordinator(
class SettingDataUpdateCoordinator( class SettingDataUpdateCoordinator(
PlenticoreUpdateCoordinator[Mapping[str, Mapping[str, str]]], PlenticoreUpdateCoordinator[Mapping[str, Mapping[str, str]]],
DataUpdateCoordinatorMixin, DataUpdateCoordinatorMixin,
): ): # pylint: disable=hass-enforce-coordinator-module
"""Implementation of PlenticoreUpdateCoordinator for settings data.""" """Implementation of PlenticoreUpdateCoordinator for settings data."""
async def _async_update_data(self) -> Mapping[str, Mapping[str, str]]: async def _async_update_data(self) -> Mapping[str, Mapping[str, str]]:
@ -237,7 +237,7 @@ class SettingDataUpdateCoordinator(
return fetched_data return fetched_data
class PlenticoreSelectUpdateCoordinator(DataUpdateCoordinator[_DataT]): class PlenticoreSelectUpdateCoordinator(DataUpdateCoordinator[_DataT]): # pylint: disable=hass-enforce-coordinator-module
"""Base implementation of DataUpdateCoordinator for Plenticore data.""" """Base implementation of DataUpdateCoordinator for Plenticore data."""
def __init__( def __init__(
@ -284,7 +284,7 @@ class PlenticoreSelectUpdateCoordinator(DataUpdateCoordinator[_DataT]):
class SelectDataUpdateCoordinator( class SelectDataUpdateCoordinator(
PlenticoreSelectUpdateCoordinator[dict[str, dict[str, str]]], PlenticoreSelectUpdateCoordinator[dict[str, dict[str, str]]],
DataUpdateCoordinatorMixin, DataUpdateCoordinatorMixin,
): ): # pylint: disable=hass-enforce-coordinator-module
"""Implementation of PlenticoreUpdateCoordinator for select data.""" """Implementation of PlenticoreUpdateCoordinator for select data."""
async def _async_update_data(self) -> dict[str, dict[str, str]]: async def _async_update_data(self) -> dict[str, dict[str, str]]:

4
homeassistant/components/melnor/models.py
View File

@ -20,7 +20,7 @@ from .const import DOMAIN
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class MelnorDataUpdateCoordinator(DataUpdateCoordinator[Device]): class MelnorDataUpdateCoordinator(DataUpdateCoordinator[Device]): # pylint: disable=hass-enforce-coordinator-module
"""Melnor data update coordinator.""" """Melnor data update coordinator."""
_device: Device _device: Device
@ -42,7 +42,7 @@ class MelnorDataUpdateCoordinator(DataUpdateCoordinator[Device]):
return self._device return self._device
class MelnorBluetoothEntity(CoordinatorEntity[MelnorDataUpdateCoordinator]): class MelnorBluetoothEntity(CoordinatorEntity[MelnorDataUpdateCoordinator]): # pylint: disable=hass-enforce-coordinator-module
"""Base class for melnor entities.""" """Base class for melnor entities."""
_device: Device _device: Device

2
homeassistant/components/mikrotik/hub.py
View File

@ -243,7 +243,7 @@ class MikrotikData:
return [] return []
class MikrotikDataUpdateCoordinator(DataUpdateCoordinator[None]): class MikrotikDataUpdateCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Mikrotik Hub Object.""" """Mikrotik Hub Object."""
def __init__( def __init__(

2
homeassistant/components/mill/__init__.py
View File

@ -21,7 +21,7 @@ _LOGGER = logging.getLogger(__name__)
PLATFORMS = [Platform.CLIMATE, Platform.SENSOR] PLATFORMS = [Platform.CLIMATE, Platform.SENSOR]
class MillDataUpdateCoordinator(DataUpdateCoordinator): class MillDataUpdateCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Mill data.""" """Class to manage fetching Mill data."""
def __init__( def __init__(

2
homeassistant/components/modern_forms/__init__.py
View File

@ -98,7 +98,7 @@ def modernforms_exception_handler(
return handler return handler
class ModernFormsDataUpdateCoordinator(DataUpdateCoordinator[ModernFormsDeviceState]): class ModernFormsDataUpdateCoordinator(DataUpdateCoordinator[ModernFormsDeviceState]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Modern Forms data from single endpoint.""" """Class to manage fetching Modern Forms data from single endpoint."""
def __init__( def __init__(

2
homeassistant/components/moehlenhoff_alpha2/__init__.py
View File

@ -52,7 +52,7 @@ async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
await hass.config_entries.async_reload(entry.entry_id) await hass.config_entries.async_reload(entry.entry_id)
class Alpha2BaseCoordinator(DataUpdateCoordinator[dict[str, dict]]): class Alpha2BaseCoordinator(DataUpdateCoordinator[dict[str, dict]]): # pylint: disable=hass-enforce-coordinator-module
"""Keep the base instance in one place and centralize the update.""" """Keep the base instance in one place and centralize the update."""
def __init__(self, hass: HomeAssistant, base: Alpha2Base) -> None: def __init__(self, hass: HomeAssistant, base: Alpha2Base) -> None:

2
homeassistant/components/nam/__init__.py
View File

@ -90,7 +90,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class NAMDataUpdateCoordinator(DataUpdateCoordinator[NAMSensors]): class NAMDataUpdateCoordinator(DataUpdateCoordinator[NAMSensors]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Nettigo Air Monitor data.""" """Class to manage fetching Nettigo Air Monitor data."""
def __init__( def __init__(

16
homeassistant/components/nextdns/__init__.py
View File

@ -47,7 +47,7 @@ from .const import (
CoordinatorDataT = TypeVar("CoordinatorDataT", bound=NextDnsData) CoordinatorDataT = TypeVar("CoordinatorDataT", bound=NextDnsData)
class NextDnsUpdateCoordinator(DataUpdateCoordinator[CoordinatorDataT]): class NextDnsUpdateCoordinator(DataUpdateCoordinator[CoordinatorDataT]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS data API.""" """Class to manage fetching NextDNS data API."""
def __init__( def __init__(
@ -84,7 +84,7 @@ class NextDnsUpdateCoordinator(DataUpdateCoordinator[CoordinatorDataT]):
raise NotImplementedError("Update method not implemented") raise NotImplementedError("Update method not implemented")
class NextDnsStatusUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsStatus]): class NextDnsStatusUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics status data from API.""" """Class to manage fetching NextDNS analytics status data from API."""
async def _async_update_data_internal(self) -> AnalyticsStatus: async def _async_update_data_internal(self) -> AnalyticsStatus:
@ -92,7 +92,7 @@ class NextDnsStatusUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsStatus]):
return await self.nextdns.get_analytics_status(self.profile_id) return await self.nextdns.get_analytics_status(self.profile_id)
class NextDnsDnssecUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsDnssec]): class NextDnsDnssecUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsDnssec]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics Dnssec data from API.""" """Class to manage fetching NextDNS analytics Dnssec data from API."""
async def _async_update_data_internal(self) -> AnalyticsDnssec: async def _async_update_data_internal(self) -> AnalyticsDnssec:
@ -100,7 +100,7 @@ class NextDnsDnssecUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsDnssec]):
return await self.nextdns.get_analytics_dnssec(self.profile_id) return await self.nextdns.get_analytics_dnssec(self.profile_id)
class NextDnsEncryptionUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsEncryption]): class NextDnsEncryptionUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsEncryption]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics encryption data from API.""" """Class to manage fetching NextDNS analytics encryption data from API."""
async def _async_update_data_internal(self) -> AnalyticsEncryption: async def _async_update_data_internal(self) -> AnalyticsEncryption:
@ -108,7 +108,7 @@ class NextDnsEncryptionUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsEncry
return await self.nextdns.get_analytics_encryption(self.profile_id) return await self.nextdns.get_analytics_encryption(self.profile_id)
class NextDnsIpVersionsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsIpVersions]): class NextDnsIpVersionsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsIpVersions]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics IP versions data from API.""" """Class to manage fetching NextDNS analytics IP versions data from API."""
async def _async_update_data_internal(self) -> AnalyticsIpVersions: async def _async_update_data_internal(self) -> AnalyticsIpVersions:
@ -116,7 +116,7 @@ class NextDnsIpVersionsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsIpVer
return await self.nextdns.get_analytics_ip_versions(self.profile_id) return await self.nextdns.get_analytics_ip_versions(self.profile_id)
class NextDnsProtocolsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsProtocols]): class NextDnsProtocolsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsProtocols]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics protocols data from API.""" """Class to manage fetching NextDNS analytics protocols data from API."""
async def _async_update_data_internal(self) -> AnalyticsProtocols: async def _async_update_data_internal(self) -> AnalyticsProtocols:
@ -124,7 +124,7 @@ class NextDnsProtocolsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsProtoc
return await self.nextdns.get_analytics_protocols(self.profile_id) return await self.nextdns.get_analytics_protocols(self.profile_id)
class NextDnsSettingsUpdateCoordinator(NextDnsUpdateCoordinator[Settings]): class NextDnsSettingsUpdateCoordinator(NextDnsUpdateCoordinator[Settings]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS connection data from API.""" """Class to manage fetching NextDNS connection data from API."""
async def _async_update_data_internal(self) -> Settings: async def _async_update_data_internal(self) -> Settings:
@ -132,7 +132,7 @@ class NextDnsSettingsUpdateCoordinator(NextDnsUpdateCoordinator[Settings]):
return await self.nextdns.get_settings(self.profile_id) return await self.nextdns.get_settings(self.profile_id)
class NextDnsConnectionUpdateCoordinator(NextDnsUpdateCoordinator[ConnectionStatus]): class NextDnsConnectionUpdateCoordinator(NextDnsUpdateCoordinator[ConnectionStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS connection data from API.""" """Class to manage fetching NextDNS connection data from API."""
async def _async_update_data_internal(self) -> ConnectionStatus: async def _async_update_data_internal(self) -> ConnectionStatus:

2
homeassistant/components/nuki/__init__.py
View File

@ -279,7 +279,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class NukiCoordinator(DataUpdateCoordinator[None]): class NukiCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Data Update Coordinator for the Nuki integration.""" """Data Update Coordinator for the Nuki integration."""
def __init__(self, hass, bridge, locks, openers): def __init__(self, hass, bridge, locks, openers):

2
homeassistant/components/nws/__init__.py
View File

@ -45,7 +45,7 @@ class NWSData:
coordinator_forecast_hourly: NwsDataUpdateCoordinator coordinator_forecast_hourly: NwsDataUpdateCoordinator
class NwsDataUpdateCoordinator(TimestampDataUpdateCoordinator[None]): class NwsDataUpdateCoordinator(TimestampDataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""NWS data update coordinator. """NWS data update coordinator.
Implements faster data update intervals for failed updates and exposes a last successful update time. Implements faster data update intervals for failed updates and exposes a last successful update time.

2
homeassistant/components/omnilogic/common.py
View File

@ -20,7 +20,7 @@ from .const import ALL_ITEM_KINDS, DOMAIN
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
class OmniLogicUpdateCoordinator(DataUpdateCoordinator[dict[tuple, dict[str, Any]]]): class OmniLogicUpdateCoordinator(DataUpdateCoordinator[dict[tuple, dict[str, Any]]]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching update data from single endpoint.""" """Class to manage fetching update data from single endpoint."""
def __init__( def __init__(

2
homeassistant/components/opengarage/__init__.py
View File

@ -50,7 +50,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class OpenGarageDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): class OpenGarageDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Opengarage data.""" """Class to manage fetching Opengarage data."""
def __init__( def __init__(

2
homeassistant/components/openweathermap/weather_update_coordinator.py
View File

@ -60,7 +60,7 @@ _LOGGER = logging.getLogger(__name__)
WEATHER_UPDATE_INTERVAL = timedelta(minutes=10) WEATHER_UPDATE_INTERVAL = timedelta(minutes=10)
class WeatherUpdateCoordinator(DataUpdateCoordinator): class WeatherUpdateCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Weather data update coordinator.""" """Weather data update coordinator."""
def __init__(self, owm, latitude, longitude, forecast_mode, hass): def __init__(self, owm, latitude, longitude, forecast_mode, hass):

2
homeassistant/components/p1_monitor/__init__.py
View File

@ -67,7 +67,7 @@ class P1MonitorData(TypedDict):
watermeter: WaterMeter | None watermeter: WaterMeter | None
class P1MonitorDataUpdateCoordinator(DataUpdateCoordinator[P1MonitorData]): class P1MonitorDataUpdateCoordinator(DataUpdateCoordinator[P1MonitorData]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching P1 Monitor data from single endpoint.""" """Class to manage fetching P1 Monitor data from single endpoint."""
config_entry: ConfigEntry config_entry: ConfigEntry

2
homeassistant/components/philips_js/__init__.py
View File

@ -85,7 +85,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class PhilipsTVDataUpdateCoordinator(DataUpdateCoordinator[None]): class PhilipsTVDataUpdateCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Coordinator to update data.""" """Coordinator to update data."""
config_entry: ConfigEntry config_entry: ConfigEntry

2
homeassistant/components/plaato/__init__.py
View File

@ -208,7 +208,7 @@ def _device_id(data):
return f"{data.get(ATTR_DEVICE_NAME)}_{data.get(ATTR_DEVICE_ID)}" return f"{data.get(ATTR_DEVICE_NAME)}_{data.get(ATTR_DEVICE_ID)}"
class PlaatoCoordinator(DataUpdateCoordinator): class PlaatoCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching data from the API.""" """Class to manage fetching data from the API."""
def __init__( def __init__(

10
homeassistant/components/prusalink/__init__.py
View File

@ -131,7 +131,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo) T = TypeVar("T", PrinterStatus, LegacyPrinterStatus, JobInfo)
class PrusaLinkUpdateCoordinator(DataUpdateCoordinator[T], ABC): class PrusaLinkUpdateCoordinator(DataUpdateCoordinator[T], ABC): # pylint: disable=hass-enforce-coordinator-module
"""Update coordinator for the printer.""" """Update coordinator for the printer."""
config_entry: ConfigEntry config_entry: ConfigEntry
@ -176,7 +176,7 @@ class PrusaLinkUpdateCoordinator(DataUpdateCoordinator[T], ABC):
return timedelta(seconds=30) return timedelta(seconds=30)
class StatusCoordinator(PrusaLinkUpdateCoordinator[PrinterStatus]): class StatusCoordinator(PrusaLinkUpdateCoordinator[PrinterStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Printer update coordinator.""" """Printer update coordinator."""
async def _fetch_data(self) -> PrinterStatus: async def _fetch_data(self) -> PrinterStatus:
@ -184,7 +184,7 @@ class StatusCoordinator(PrusaLinkUpdateCoordinator[PrinterStatus]):
return await self.api.get_status() return await self.api.get_status()
class LegacyStatusCoordinator(PrusaLinkUpdateCoordinator[LegacyPrinterStatus]): class LegacyStatusCoordinator(PrusaLinkUpdateCoordinator[LegacyPrinterStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Printer legacy update coordinator.""" """Printer legacy update coordinator."""
async def _fetch_data(self) -> LegacyPrinterStatus: async def _fetch_data(self) -> LegacyPrinterStatus:
@ -192,7 +192,7 @@ class LegacyStatusCoordinator(PrusaLinkUpdateCoordinator[LegacyPrinterStatus]):
return await self.api.get_legacy_printer() return await self.api.get_legacy_printer()
class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]): class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]): # pylint: disable=hass-enforce-coordinator-module
"""Job update coordinator.""" """Job update coordinator."""
async def _fetch_data(self) -> JobInfo: async def _fetch_data(self) -> JobInfo:
@ -200,7 +200,7 @@ class JobUpdateCoordinator(PrusaLinkUpdateCoordinator[JobInfo]):
return await self.api.get_job() return await self.api.get_job()
class PrusaLinkEntity(CoordinatorEntity[PrusaLinkUpdateCoordinator]): class PrusaLinkEntity(CoordinatorEntity[PrusaLinkUpdateCoordinator]): # pylint: disable=hass-enforce-coordinator-module
"""Defines a base PrusaLink entity.""" """Defines a base PrusaLink entity."""
_attr_has_entity_name = True _attr_has_entity_name = True

2
homeassistant/components/pure_energie/__init__.py
View File

@ -47,7 +47,7 @@ class PureEnergieData(NamedTuple):
smartbridge: SmartBridge smartbridge: SmartBridge
class PureEnergieDataUpdateCoordinator(DataUpdateCoordinator[PureEnergieData]): class PureEnergieDataUpdateCoordinator(DataUpdateCoordinator[PureEnergieData]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Pure Energie data from single eindpoint.""" """Class to manage fetching Pure Energie data from single eindpoint."""
config_entry: ConfigEntry config_entry: ConfigEntry

2
homeassistant/components/pvpc_hourly_pricing/__init__.py
View File

@ -59,7 +59,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class ElecPricesDataUpdateCoordinator(DataUpdateCoordinator[EsiosApiData]): class ElecPricesDataUpdateCoordinator(DataUpdateCoordinator[EsiosApiData]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Electricity prices data from API.""" """Class to manage fetching Electricity prices data from API."""
def __init__( def __init__(

2
homeassistant/components/rainforest_eagle/data.py
View File

@ -87,7 +87,7 @@ async def async_get_type(hass, cloud_id, install_code, host):
return None, None return None, None
class EagleDataCoordinator(DataUpdateCoordinator): class EagleDataCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Get the latest data from the Eagle device.""" """Get the latest data from the Eagle device."""
eagle100_reader: Eagle100Reader | None = None eagle100_reader: Eagle100Reader | None = None

2
homeassistant/components/rainmachine/util.py
View File

@ -84,7 +84,7 @@ def key_exists(data: dict[str, Any], search_key: str) -> bool:
return False return False
class RainMachineDataUpdateCoordinator(DataUpdateCoordinator[dict]): class RainMachineDataUpdateCoordinator(DataUpdateCoordinator[dict]): # pylint: disable=hass-enforce-coordinator-module
"""Define an extended DataUpdateCoordinator.""" """Define an extended DataUpdateCoordinator."""
config_entry: ConfigEntry config_entry: ConfigEntry

4
homeassistant/components/risco/__init__.py
View File

@ -197,7 +197,7 @@ async def _update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
await hass.config_entries.async_reload(entry.entry_id) await hass.config_entries.async_reload(entry.entry_id)
class RiscoDataUpdateCoordinator(DataUpdateCoordinator[Alarm]): class RiscoDataUpdateCoordinator(DataUpdateCoordinator[Alarm]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching risco data.""" """Class to manage fetching risco data."""
def __init__( def __init__(
@ -221,7 +221,7 @@ class RiscoDataUpdateCoordinator(DataUpdateCoordinator[Alarm]):
raise UpdateFailed(error) from error raise UpdateFailed(error) from error
class RiscoEventsDataUpdateCoordinator(DataUpdateCoordinator[list[Event]]): class RiscoEventsDataUpdateCoordinator(DataUpdateCoordinator[list[Event]]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching risco data.""" """Class to manage fetching risco data."""
def __init__( def __init__(

2
homeassistant/components/sharkiq/update_coordinator.py
View File

@ -20,7 +20,7 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
from .const import API_TIMEOUT, DOMAIN, LOGGER, UPDATE_INTERVAL from .const import API_TIMEOUT, DOMAIN, LOGGER, UPDATE_INTERVAL
class SharkIqUpdateCoordinator(DataUpdateCoordinator[bool]): class SharkIqUpdateCoordinator(DataUpdateCoordinator[bool]): # pylint: disable=hass-enforce-coordinator-module
"""Define a wrapper class to update Shark IQ data.""" """Define a wrapper class to update Shark IQ data."""
def __init__( def __init__(

2
homeassistant/components/surepetcare/__init__.py
View File

@ -102,7 +102,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class SurePetcareDataCoordinator(DataUpdateCoordinator[dict[int, SurepyEntity]]): class SurePetcareDataCoordinator(DataUpdateCoordinator[dict[int, SurepyEntity]]): # pylint: disable=hass-enforce-coordinator-module
"""Handle Surepetcare data.""" """Handle Surepetcare data."""
def __init__(self, entry: ConfigEntry, hass: HomeAssistant) -> None: def __init__(self, entry: ConfigEntry, hass: HomeAssistant) -> None:

2
homeassistant/components/switcher_kis/__init__.py
View File

@ -125,7 +125,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
class SwitcherDataUpdateCoordinator( class SwitcherDataUpdateCoordinator(
update_coordinator.DataUpdateCoordinator[SwitcherBase] update_coordinator.DataUpdateCoordinator[SwitcherBase]
): ): # pylint: disable=hass-enforce-coordinator-module
"""Switcher device data update coordinator.""" """Switcher device data update coordinator."""
def __init__( def __init__(

4
homeassistant/components/tibber/sensor.py
View File

@ -498,7 +498,7 @@ class TibberSensorRT(TibberSensor, CoordinatorEntity["TibberRtDataCoordinator"])
self.async_write_ha_state() self.async_write_ha_state()
class TibberRtDataCoordinator(DataUpdateCoordinator): class TibberRtDataCoordinator(DataUpdateCoordinator): # pylint: disable=hass-enforce-coordinator-module
"""Handle Tibber realtime data.""" """Handle Tibber realtime data."""
def __init__( def __init__(
@ -562,7 +562,7 @@ class TibberRtDataCoordinator(DataUpdateCoordinator):
return self.data.get("data", {}).get("liveMeasurement") return self.data.get("data", {}).get("liveMeasurement")
class TibberDataCoordinator(DataUpdateCoordinator[None]): class TibberDataCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Handle Tibber data and insert statistics.""" """Handle Tibber data and insert statistics."""
config_entry: ConfigEntry config_entry: ConfigEntry

4
homeassistant/components/tolo/__init__.py
View File

@ -63,7 +63,7 @@ class ToloSaunaData(NamedTuple):
settings: SettingsInfo settings: SettingsInfo
class ToloSaunaUpdateCoordinator(DataUpdateCoordinator[ToloSaunaData]): class ToloSaunaUpdateCoordinator(DataUpdateCoordinator[ToloSaunaData]): # pylint: disable=hass-enforce-coordinator-module
"""DataUpdateCoordinator for TOLO Sauna.""" """DataUpdateCoordinator for TOLO Sauna."""
def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None: def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
@ -92,7 +92,7 @@ class ToloSaunaUpdateCoordinator(DataUpdateCoordinator[ToloSaunaData]):
return ToloSaunaData(status, settings) return ToloSaunaData(status, settings)
class ToloSaunaCoordinatorEntity(CoordinatorEntity[ToloSaunaUpdateCoordinator]): class ToloSaunaCoordinatorEntity(CoordinatorEntity[ToloSaunaUpdateCoordinator]): # pylint: disable=hass-enforce-coordinator-module
"""CoordinatorEntity for TOLO Sauna.""" """CoordinatorEntity for TOLO Sauna."""
_attr_has_entity_name = True _attr_has_entity_name = True

2
homeassistant/components/tomorrowio/__init__.py
View File

@ -163,7 +163,7 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
return unload_ok return unload_ok
class TomorrowioDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): class TomorrowioDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): # pylint: disable=hass-enforce-coordinator-module
"""Define an object to hold Tomorrow.io data.""" """Define an object to hold Tomorrow.io data."""
def __init__(self, hass: HomeAssistant, api: TomorrowioV4) -> None: def __init__(self, hass: HomeAssistant, api: TomorrowioV4) -> None:

2
homeassistant/components/totalconnect/__init__.py
View File

@ -78,7 +78,7 @@ async def update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
client.locations[location_id].auto_bypass_low_battery = bypass client.locations[location_id].auto_bypass_low_battery = bypass
class TotalConnectDataUpdateCoordinator(DataUpdateCoordinator[None]): class TotalConnectDataUpdateCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Class to fetch data from TotalConnect.""" """Class to fetch data from TotalConnect."""
config_entry: ConfigEntry config_entry: ConfigEntry

4
homeassistant/components/tplink_omada/controller.py
View File

@ -15,7 +15,7 @@ POLL_SWITCH_PORT = 300
POLL_GATEWAY = 300 POLL_GATEWAY = 300
class OmadaSwitchPortCoordinator(OmadaCoordinator[OmadaSwitchPortDetails]): class OmadaSwitchPortCoordinator(OmadaCoordinator[OmadaSwitchPortDetails]): # pylint: disable=hass-enforce-coordinator-module
"""Coordinator for getting details about ports on a switch.""" """Coordinator for getting details about ports on a switch."""
def __init__( def __init__(
@ -36,7 +36,7 @@ class OmadaSwitchPortCoordinator(OmadaCoordinator[OmadaSwitchPortDetails]):
return {p.port_id: p for p in ports} return {p.port_id: p for p in ports}
class OmadaGatewayCoordinator(OmadaCoordinator[OmadaGateway]): class OmadaGatewayCoordinator(OmadaCoordinator[OmadaGateway]): # pylint: disable=hass-enforce-coordinator-module
"""Coordinator for getting details about the site's gateway.""" """Coordinator for getting details about the site's gateway."""
def __init__( def __init__(

2
homeassistant/components/tplink_omada/update.py
View File

@ -34,7 +34,7 @@ class FirmwareUpdateStatus(NamedTuple):
firmware: OmadaFirmwareUpdate | None firmware: OmadaFirmwareUpdate | None
class OmadaFirmwareUpdateCoodinator(OmadaCoordinator[FirmwareUpdateStatus]): class OmadaFirmwareUpdateCoodinator(OmadaCoordinator[FirmwareUpdateStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Coordinator for getting details about ports on a switch.""" """Coordinator for getting details about ports on a switch."""
def __init__(self, hass: HomeAssistant, omada_client: OmadaSiteClient) -> None: def __init__(self, hass: HomeAssistant, omada_client: OmadaSiteClient) -> None:

2
homeassistant/components/ukraine_alarm/__init__.py
View File

@ -46,7 +46,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class UkraineAlarmDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): class UkraineAlarmDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Ukraine Alarm API.""" """Class to manage fetching Ukraine Alarm API."""
def __init__( def __init__(

2
homeassistant/components/upcloud/__init__.py
View File

@ -57,7 +57,7 @@ STATE_MAP = {"error": STATE_PROBLEM, "started": STATE_ON, "stopped": STATE_OFF}
class UpCloudDataUpdateCoordinator( class UpCloudDataUpdateCoordinator(
DataUpdateCoordinator[dict[str, upcloud_api.Server]] DataUpdateCoordinator[dict[str, upcloud_api.Server]]
): ): # pylint: disable=hass-enforce-coordinator-module
"""UpCloud data update coordinator.""" """UpCloud data update coordinator."""
def __init__( def __init__(

2
homeassistant/components/vallox/__init__.py
View File

@ -155,7 +155,7 @@ class ValloxState:
return next_filter_change_date return next_filter_change_date
class ValloxDataUpdateCoordinator(DataUpdateCoordinator[ValloxState]): class ValloxDataUpdateCoordinator(DataUpdateCoordinator[ValloxState]): # pylint: disable=hass-enforce-coordinator-module
"""The DataUpdateCoordinator for Vallox.""" """The DataUpdateCoordinator for Vallox."""

2
homeassistant/components/venstar/__init__.py
View File

@ -64,7 +64,7 @@ async def async_unload_entry(hass: HomeAssistant, config: ConfigEntry) -> bool:
return unload_ok return unload_ok
class VenstarDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator[None]): class VenstarDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Venstar data.""" """Class to manage fetching Venstar data."""
def __init__( def __init__(

2
homeassistant/components/vizio/__init__.py
View File

@ -97,7 +97,7 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
return unload_ok return unload_ok
class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]): class VizioAppsDataUpdateCoordinator(DataUpdateCoordinator[list[dict[str, Any]]]): # pylint: disable=hass-enforce-coordinator-module
"""Define an object to hold Vizio app config data.""" """Define an object to hold Vizio app config data."""
def __init__(self, hass: HomeAssistant, store: Store[list[dict[str, Any]]]) -> None: def __init__(self, hass: HomeAssistant, store: Store[list[dict[str, Any]]]) -> None:

2
homeassistant/components/volvooncall/__init__.py
View File

@ -168,7 +168,7 @@ class VolvoData:
raise InvalidAuth from exc raise InvalidAuth from exc
class VolvoUpdateCoordinator(DataUpdateCoordinator[None]): class VolvoUpdateCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Volvo coordinator.""" """Volvo coordinator."""
def __init__(self, hass: HomeAssistant, volvo_data: VolvoData) -> None: def __init__(self, hass: HomeAssistant, volvo_data: VolvoData) -> None:

2
homeassistant/components/wemo/wemo_device.py
View File

@ -85,7 +85,7 @@ class Options:
) )
class DeviceCoordinator(DataUpdateCoordinator[None]): class DeviceCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Home Assistant wrapper for a pyWeMo device.""" """Home Assistant wrapper for a pyWeMo device."""
options: Options | None = None options: Options | None = None

2
homeassistant/components/xbox/__init__.py
View File

@ -123,7 +123,7 @@ class XboxData:
presence: dict[str, PresenceData] presence: dict[str, PresenceData]
class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]): class XboxUpdateCoordinator(DataUpdateCoordinator[XboxData]): # pylint: disable=hass-enforce-coordinator-module
"""Store Xbox Console Status.""" """Store Xbox Console Status."""
def __init__( def __init__(

2
homeassistant/components/yamaha_musiccast/__init__.py
View File

@ -104,7 +104,7 @@ async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None:
await hass.config_entries.async_reload(entry.entry_id) await hass.config_entries.async_reload(entry.entry_id)
class MusicCastDataUpdateCoordinator(DataUpdateCoordinator[MusicCastData]): class MusicCastDataUpdateCoordinator(DataUpdateCoordinator[MusicCastData]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching data from the API.""" """Class to manage fetching data from the API."""
def __init__(self, hass: HomeAssistant, client: MusicCastDevice) -> None: def __init__(self, hass: HomeAssistant, client: MusicCastDevice) -> None:

54
pylint/plugins/hass_enforce_coordinator_module.py Normal file
View File

@ -0,0 +1,54 @@
"""Plugin for checking if coordinator is in its own module."""
from __future__ import annotations
from astroid import nodes
from pylint.checkers import BaseChecker
from pylint.lint import PyLinter
class HassEnforceCoordinatorModule(BaseChecker):
"""Checker for coordinators own module."""
name = "hass_enforce_coordinator_module"
priority = -1
msgs = {
"C7461": (
"Derived data update coordinator is recommended to be placed in the 'coordinator' module",
"hass-enforce-coordinator-module",
"Used when derived data update coordinator should be placed in its own module.",
),
}
options = (
(
"ignore-wrong-coordinator-module",
{
"default": False,
"type": "yn",
"metavar": "<y or n>",
"help": "Set to ``no`` if you wish to check if derived data update coordinator "
"is placed in its own module.",
},
),
)
def visit_classdef(self, node: nodes.ClassDef) -> None:
"""Check if derived data update coordinator is placed in its own module."""
if self.linter.config.ignore_wrong_coordinator_module:
return
root_name = node.root().name
# we only want to check component update coordinators
if not root_name.startswith("homeassistant.components"):
return
is_coordinator_module = root_name.endswith(".coordinator")
for ancestor in node.ancestors():
if ancestor.name == "DataUpdateCoordinator" and not is_coordinator_module:
self.add_message("hass-enforce-coordinator-module", node=node)
return
def register(linter: PyLinter) -> None:
"""Register the checker."""
linter.register_checker(HassEnforceCoordinatorModule(linter))

1
pyproject.toml
View File

@ -103,6 +103,7 @@ init-hook = """\
load-plugins = [ load-plugins = [
"pylint.extensions.code_style", "pylint.extensions.code_style",
"pylint.extensions.typing", "pylint.extensions.typing",
"hass_enforce_coordinator_module",
"hass_enforce_sorted_platforms", "hass_enforce_sorted_platforms",
"hass_enforce_super_call", "hass_enforce_super_call",
"hass_enforce_type_hints", "hass_enforce_type_hints",

21
tests/pylint/conftest.py
View File

@ -101,3 +101,24 @@ def enforce_sorted_platforms_checker_fixture(
) )
enforce_sorted_platforms_checker.module = "homeassistant.components.pylint_test" enforce_sorted_platforms_checker.module = "homeassistant.components.pylint_test"
return enforce_sorted_platforms_checker return enforce_sorted_platforms_checker
@pytest.fixture(name="hass_enforce_coordinator_module", scope="session")
def hass_enforce_coordinator_module_fixture() -> ModuleType:
"""Fixture to the content for the hass_enforce_coordinator_module check."""
return _load_plugin_from_file(
"hass_enforce_coordinator_module",
"pylint/plugins/hass_enforce_coordinator_module.py",
)
@pytest.fixture(name="enforce_coordinator_module_checker")
def enforce_coordinator_module_fixture(
hass_enforce_coordinator_module, linter
) -> BaseChecker:
"""Fixture to provide a hass_enforce_coordinator_module checker."""
enforce_coordinator_module_checker = (
hass_enforce_coordinator_module.HassEnforceCoordinatorModule(linter)
)
enforce_coordinator_module_checker.module = "homeassistant.components.pylint_test"
return enforce_coordinator_module_checker

133
tests/pylint/test_enforce_coordinator_module.py Normal file
View File

@ -0,0 +1,133 @@
"""Tests for pylint hass_enforce_coordinator_module plugin."""
from __future__ import annotations
import astroid
from pylint.checkers import BaseChecker
from pylint.interfaces import UNDEFINED
from pylint.testutils import MessageTest
from pylint.testutils.unittest_linter import UnittestLinter
from pylint.utils.ast_walker import ASTWalker
import pytest
from . import assert_adds_messages, assert_no_messages
@pytest.mark.parametrize(
"code",
[
pytest.param(
"""
class DataUpdateCoordinator:
pass
class TestCoordinator(DataUpdateCoordinator):
pass
""",
id="simple",
),
pytest.param(
"""
class DataUpdateCoordinator:
pass
class TestCoordinator(DataUpdateCoordinator):
pass
class TestCoordinator2(TestCoordinator):
pass
""",
id="nested",
),
],
)
def test_enforce_coordinator_module_good(
linter: UnittestLinter, enforce_coordinator_module_checker: BaseChecker, code: str
) -> None:
"""Good test cases."""
root_node = astroid.parse(code, "homeassistant.components.pylint_test.coordinator")
walker = ASTWalker(linter)
walker.add_checker(enforce_coordinator_module_checker)
with assert_no_messages(linter):
walker.walk(root_node)
def test_enforce_coordinator_module_bad_simple(
linter: UnittestLinter,
enforce_coordinator_module_checker: BaseChecker,
) -> None:
"""Bad test case with coordinator extending directly."""
root_node = astroid.parse(
"""
class DataUpdateCoordinator:
pass
class TestCoordinator(DataUpdateCoordinator):
pass
""",
"homeassistant.components.pylint_test",
)
walker = ASTWalker(linter)
walker.add_checker(enforce_coordinator_module_checker)
with assert_adds_messages(
linter,
MessageTest(
msg_id="hass-enforce-coordinator-module",
line=5,
node=root_node.body[1],
args=None,
confidence=UNDEFINED,
col_offset=0,
end_line=5,
end_col_offset=21,
),
):
walker.walk(root_node)
def test_enforce_coordinator_module_bad_nested(
linter: UnittestLinter,
enforce_coordinator_module_checker: BaseChecker,
) -> None:
"""Bad test case with nested coordinators."""
root_node = astroid.parse(
"""
class DataUpdateCoordinator:
pass
class TestCoordinator(DataUpdateCoordinator):
pass
class NopeCoordinator(TestCoordinator):
pass
""",
"homeassistant.components.pylint_test",
)
walker = ASTWalker(linter)
walker.add_checker(enforce_coordinator_module_checker)
with assert_adds_messages(
linter,
MessageTest(
msg_id="hass-enforce-coordinator-module",
line=5,
node=root_node.body[1],
args=None,
confidence=UNDEFINED,
col_offset=0,
end_line=5,
end_col_offset=21,
),
MessageTest(
msg_id="hass-enforce-coordinator-module",
line=8,
node=root_node.body[2],
args=None,
confidence=UNDEFINED,
col_offset=0,
end_line=8,
end_col_offset=21,
),
):
walker.walk(root_node)