Clean up Mealie coordinator (#122310)

* Clean up Mealie coordinator

* Clean up Mealie coordinator

* Clean up Mealie coordinator

* Fix

* Fix
This commit is contained in:
Joost Lekkerkerker 2024-07-21 16:24:46 +02:00 committed by GitHub
parent a78d6b8c36
commit 8d01ad98eb
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 55 additions and 72 deletions

View file

@ -2,6 +2,7 @@
from __future__ import annotations
from abc import abstractmethod
from dataclasses import dataclass
from datetime import timedelta
@ -44,56 +45,50 @@ class MealieDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
"""Base coordinator."""
config_entry: MealieConfigEntry
_name: str
_update_interval: timedelta
def __init__(
self,
hass: HomeAssistant,
name: str,
client: MealieClient,
update_interval: timedelta,
) -> None:
"""Initialize the Withings data coordinator."""
def __init__(self, hass: HomeAssistant, client: MealieClient) -> None:
"""Initialize the Mealie data coordinator."""
super().__init__(
hass,
LOGGER,
name=name,
update_interval=update_interval,
name=self._name,
update_interval=self._update_interval,
)
self.client = client
async def _async_update_data(self) -> _DataT:
"""Fetch data from Mealie."""
try:
return await self._async_update_internal()
except MealieAuthenticationError as error:
raise ConfigEntryAuthFailed from error
except MealieConnectionError as error:
raise UpdateFailed(error) from error
@abstractmethod
async def _async_update_internal(self) -> _DataT:
"""Fetch data from Mealie."""
class MealieMealplanCoordinator(
MealieDataUpdateCoordinator[dict[MealplanEntryType, list[Mealplan]]]
):
"""Class to manage fetching Mealie data."""
def __init__(self, hass: HomeAssistant, client: MealieClient) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
name="MealieMealplan",
client=client,
update_interval=timedelta(hours=1),
)
self.client = client
_name = "MealieMealplan"
_update_interval = timedelta(hours=1)
async def _async_update_data(self) -> dict[MealplanEntryType, list[Mealplan]]:
async def _async_update_internal(self) -> dict[MealplanEntryType, list[Mealplan]]:
next_week = dt_util.now() + WEEK
try:
data = (
await self.client.get_mealplans(dt_util.now().date(), next_week.date())
).items
except MealieAuthenticationError as error:
raise ConfigEntryAuthFailed from error
except MealieConnectionError as error:
raise UpdateFailed(error) from error
current_date = dt_util.now().date()
next_week_date = next_week.date()
response = await self.client.get_mealplans(current_date, next_week_date)
res: dict[MealplanEntryType, list[Mealplan]] = {
MealplanEntryType.BREAKFAST: [],
MealplanEntryType.LUNCH: [],
MealplanEntryType.DINNER: [],
MealplanEntryType.SIDE: [],
type_: [] for type_ in MealplanEntryType
}
for meal in data:
for meal in response.items:
res[meal.entry_type].append(meal)
return res
@ -111,56 +106,34 @@ class MealieShoppingListCoordinator(
):
"""Class to manage fetching Mealie Shopping list data."""
def __init__(self, hass: HomeAssistant, client: MealieClient) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
name="MealieShoppingLists",
client=client,
update_interval=timedelta(minutes=5),
)
_name = "MealieShoppingList"
_update_interval = timedelta(minutes=5)
async def _async_update_data(
async def _async_update_internal(
self,
) -> dict[str, ShoppingListData]:
shopping_list_items = {}
try:
shopping_lists = (await self.client.get_shopping_lists()).items
for shopping_list in shopping_lists:
shopping_list_id = shopping_list.list_id
shopping_lists = (await self.client.get_shopping_lists()).items
for shopping_list in shopping_lists:
shopping_list_id = shopping_list.list_id
shopping_items = (
await self.client.get_shopping_items(shopping_list_id)
).items
shopping_items = (
await self.client.get_shopping_items(shopping_list_id)
).items
shopping_list_items[shopping_list_id] = ShoppingListData(
shopping_list=shopping_list, items=shopping_items
)
except MealieAuthenticationError as error:
raise ConfigEntryAuthFailed from error
except MealieConnectionError as error:
raise UpdateFailed(error) from error
shopping_list_items[shopping_list_id] = ShoppingListData(
shopping_list=shopping_list, items=shopping_items
)
return shopping_list_items
class MealieStatisticsCoordinator(MealieDataUpdateCoordinator[Statistics]):
"""Class to manage fetching Mealie Statistics data."""
def __init__(self, hass: HomeAssistant, client: MealieClient) -> None:
"""Initialize coordinator."""
super().__init__(
hass,
name="MealieStatistics",
client=client,
update_interval=timedelta(minutes=15),
)
_name = "MealieStatistics"
_update_interval = timedelta(minutes=15)
async def _async_update_data(
async def _async_update_internal(
self,
) -> Statistics:
try:
return await self.client.get_statistics()
except MealieAuthenticationError as error:
raise ConfigEntryAuthFailed from error
except MealieConnectionError as error:
raise UpdateFailed(error) from error
return await self.client.get_statistics()

View file

@ -32,6 +32,15 @@ async def test_device_info(
assert device_entry == snapshot
@pytest.mark.parametrize(
"field",
[
"get_about",
"get_mealplans",
"get_shopping_lists",
"get_statistics",
],
)
@pytest.mark.parametrize(
("exc", "state"),
[
@ -43,11 +52,12 @@ async def test_setup_failure(
hass: HomeAssistant,
mock_mealie_client: AsyncMock,
mock_config_entry: MockConfigEntry,
field: str,
exc: Exception,
state: ConfigEntryState,
) -> None:
"""Test setup failure."""
mock_mealie_client.get_about.side_effect = exc
getattr(mock_mealie_client, field).side_effect = exc
await setup_integration(hass, mock_config_entry)