starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-08-24 10:13:52 -05:00
Code Issues Packages Projects Releases Wiki Activity

Refactor coordinator in Steam integration (#174661)

Browse Source
This commit is contained in:
Manu
2026-06-30 16:31:21 +02:00
committed by GitHub
parent 5a1cc024dd
commit 1a330ca23e
8 changed files with 139 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/steam_online/config_flow.py
+9 -6
View File
@@ -1,6 +1,7 @@
"""Config flow for Steam integration."""
from collections.abc import Iterator, Mapping
import logging
from typing import Any, override
import steam.api
@@ -17,9 +18,12 @@ from homeassistant.const import CONF_API_KEY, CONF_NAME, Platform
from homeassistant.core import callback
from homeassistant.helpers import config_validation as cv, entity_registry as er
from .const import CONF_ACCOUNT, CONF_ACCOUNTS, DOMAIN, LOGGER, PLACEHOLDERS
from .const import CONF_ACCOUNT, CONF_ACCOUNTS, DOMAIN, PLACEHOLDERS
from .coordinator import SteamConfigEntry
_LOGGER = logging.getLogger(__name__)
# To avoid too long request URIs, the amount of ids to request is limited
MAX_IDS_TO_REQUEST = 275
@@ -75,8 +79,8 @@ class SteamFlowHandler(ConfigFlow, domain=DOMAIN):
errors["base"] = (
"invalid_auth" if "403" in str(ex) else "cannot_connect"
)
except Exception: # noqa: BLE001
LOGGER.exception("Unknown exception")
except Exception:
_LOGGER.exception("Unknown exception")
errors["base"] = "unknown"
if not errors:
return self.async_create_entry(
@@ -129,8 +133,8 @@ class SteamFlowHandler(ConfigFlow, domain=DOMAIN):
errors["base"] = (
"invalid_auth" if "403" in str(ex) else "cannot_connect"
)
except Exception: # noqa: BLE001
LOGGER.exception("Unknown exception")
except Exception:
_LOGGER.exception("Unknown exception")
errors["base"] = "unknown"
if not errors:
@@ -166,7 +170,6 @@ class SteamOptionsFlowHandler(OptionsFlowWithReload):
) -> ConfigFlowResult:
"""Manage Steam options."""
if user_input is not None:
await self.hass.config_entries.async_unload(self.config_entry.entry_id)
for _id in self.options[CONF_ACCOUNTS]:
if _id not in user_input[CONF_ACCOUNTS] and (
entity_id := er.async_get(self.hass).async_get_entity_id(
homeassistant/components/steam_online/const.py
-2
View File
@@ -1,6 +1,5 @@
"""Steam constants."""
import logging
from typing import Final
CONF_ACCOUNT = "account"
@@ -10,7 +9,6 @@ DATA_KEY_COORDINATOR = "coordinator"
DEFAULT_NAME = "Steam"
DOMAIN: Final = "steam_online"
LOGGER = logging.getLogger(__package__)
PLACEHOLDERS = {
"api_key_url": "https://steamcommunity.com/dev/apikey",
homeassistant/components/steam_online/coordinator.py
+82 -30
View File
@@ -1,10 +1,11 @@
"""Data update coordinator for the Steam integration."""
from dataclasses import dataclass
from datetime import timedelta
import logging
from typing import override
import steam.api
from steam.api import _interface_method as INTMethod
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY
@@ -12,65 +13,116 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import CONF_ACCOUNTS, DOMAIN, LOGGER
from .const import CONF_ACCOUNTS, DOMAIN
type SteamConfigEntry = ConfigEntry[SteamDataUpdateCoordinator]
_LOGGER = logging.getLogger(__name__)
class SteamDataUpdateCoordinator(
DataUpdateCoordinator[dict[str, dict[str, str | int]]]
):
@dataclass(kw_only=True, frozen=True)
class PlayerData:
"""Steam player data."""
steamid: str
communityvisibilitystate: int
profilestate: int
personaname: str
commentpermission: int | None = None
profileurl: str
avatar: str
avatarmedium: str
avatarfull: str
avatarhash: str
lastlogoff: int
personastate: int
realname: str | None = None
primaryclanid: str | None = None
timecreated: int | None = None
personastateflags: int
loccountrycode: str | None = None
locstatecode: str | None = None
loccityid: int | None = None
gameextrainfo: str | None = None
gameid: str | None = None
level: int | None = None
class SteamDataUpdateCoordinator(DataUpdateCoordinator[dict[str, PlayerData]]):
"""Data update coordinator for the Steam integration."""
config_entry: SteamConfigEntry
user_interface: steam.api.interface
player_interface: steam.api.interface
def __init__(self, hass: HomeAssistant, config_entry: SteamConfigEntry) -> None:
"""Initialize the coordinator."""
super().__init__(
hass=hass,
logger=LOGGER,
logger=_LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=timedelta(seconds=30),
)
self.game_icons: dict[int, str] = {}
self.player_interface: INTMethod = None
self.user_interface: INTMethod = None
steam.api.key.set(self.config_entry.data[CONF_API_KEY])
self.game_icons: dict[str, str] = {}
def _update(self) -> dict[str, dict[str, str | int]]:
@override
async def _async_setup(self) -> None:
"""Set up the coordinator."""
steam.api.key.set(self.config_entry.data[CONF_API_KEY])
self.user_interface = steam.api.interface("ISteamUser")
self.player_interface = steam.api.interface("IPlayerService")
def _update(self) -> dict[str, PlayerData]:
"""Fetch data from API endpoint."""
accounts = self.config_entry.options[CONF_ACCOUNTS]
_ids = list(accounts)
if not self.user_interface or not self.player_interface:
self.user_interface = steam.api.interface("ISteamUser")
self.player_interface = steam.api.interface("IPlayerService")
if not self.game_icons:
for _id in _ids:
res = self.player_interface.GetOwnedGames(
steamid=_id, include_appinfo=1
)["response"]
self.game_icons = self.game_icons | {
game["appid"]: game["img_icon_url"] for game in res.get("games", [])
}
response = self.user_interface.GetPlayerSummaries(steamids=_ids)
players = {
player["steamid"]: player
player["steamid"]: PlayerData(
**player,
level=self.player_interface.GetSteamLevel(steamid=player["steamid"])[
"response"
].get("player_level"),
)
for player in response["response"]["players"]["player"]
if player["steamid"] in _ids
}
for value in players.values():
data = self.player_interface.GetSteamLevel(steamid=value["steamid"])
value["level"] = data["response"].get("player_level")
for player in players.values():
if player.gameid and player.gameid not in self.game_icons:
games = self.player_interface.GetOwnedGames(
steamid=player.steamid,
include_appinfo=1,
include_played_free_games=True,
)["response"].get("games", [])
self.game_icons.update(
{str(game["appid"]): game["img_icon_url"] for game in games}
)
return players
@override
async def _async_update_data(self) -> dict[str, dict[str, str | int]]:
async def _async_update_data(self) -> dict[str, PlayerData]:
"""Send request to the executor."""
try:
return await self.hass.async_add_executor_job(self._update)
except steam.api.HTTPTimeoutError as ex:
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="timeout_exception",
) from ex
except steam.api.HTTPError as ex:
if "401" in str(ex):
raise ConfigEntryAuthFailed from ex
raise UpdateFailed(ex) from ex
_LOGGER.debug("Full exception:", exc_info=True)
if "401" in str(ex) or "403" in str(ex):
raise ConfigEntryAuthFailed(
translation_domain=DOMAIN,
translation_key="auth_exception",
) from ex
raise UpdateFailed(
translation_domain=DOMAIN,
translation_key="request_exception",
) from ex
homeassistant/components/steam_online/sensor.py
+16 -18
View File
@@ -4,7 +4,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from enum import StrEnum
from typing import Any, cast, override
from typing import Any, override
from homeassistant.components.sensor import SensorEntity, SensorEntityDescription
from homeassistant.core import HomeAssistant
@@ -20,7 +20,7 @@ from .const import (
STEAM_MAIN_IMAGE_FILE,
STEAM_STATUSES,
)
from .coordinator import SteamConfigEntry, SteamDataUpdateCoordinator
from .coordinator import PlayerData, SteamConfigEntry, SteamDataUpdateCoordinator
from .entity import SteamEntity
PARALLEL_UPDATES = 1
@@ -36,18 +36,18 @@ class SteamSensor(StrEnum):
class SteamSensorEntityDescription(SensorEntityDescription):
"""Steam sensor description."""
value_fn: Callable[[dict[str, Any]], StateType]
name_fn: Callable[[dict[str, Any]], str]
entity_picture_fn: Callable[[dict[str, Any]], str] | None = None
value_fn: Callable[[PlayerData], StateType]
name_fn: Callable[[PlayerData], str]
entity_picture_fn: Callable[[PlayerData], str] | None = None
SENSOR_DESCRIPTIONS: tuple[SteamSensorEntityDescription, ...] = (
SteamSensorEntityDescription(
key=SteamSensor.ACCOUNT,
translation_key=SteamSensor.ACCOUNT,
value_fn=lambda x: STEAM_STATUSES[x["personastate"]],
name_fn=lambda x: x["personaname"],
entity_picture_fn=lambda x: x["avatarfull"],
value_fn=lambda x: STEAM_STATUSES[x.personastate],
name_fn=lambda x: x.personaname,
entity_picture_fn=lambda x: x.avatarfull,
),
)
@@ -106,29 +106,27 @@ class SteamSensorEntity(SteamEntity, SensorEntity):
player = self.coordinator.data[self._steamid]
attrs: dict[str, str | int | datetime] = {}
if game := player.get("gameextrainfo"):
if game := player.gameextrainfo:
attrs["game"] = game
if game_id := player.get("gameid"):
if game_id := player.gameid:
attrs["game_id"] = game_id
game_url = f"{STEAM_API_URL}{player['gameid']}/"
game_url = f"{STEAM_API_URL}{player.gameid}/"
attrs["game_image_header"] = f"{game_url}{STEAM_HEADER_IMAGE_FILE}"
attrs["game_image_main"] = f"{game_url}{STEAM_MAIN_IMAGE_FILE}"
if info := self._get_game_icon(player):
attrs["game_icon"] = f"{STEAM_ICON_URL}{game_id}/{info}.jpg"
if last_online := cast(int | None, player.get("lastlogoff")):
if last_online := player.lastlogoff:
attrs["last_online"] = dt_util.as_local(
dt_util.utc_from_timestamp(last_online)
)
if level := self.coordinator.data[self._steamid]["level"]:
if level := self.coordinator.data[self._steamid].level:
attrs["level"] = level
return attrs
def _get_game_icon(self, player: dict) -> str | None:
def _get_game_icon(self, player: PlayerData) -> str | None:
"""Get game icon identifier."""
if player.get("gameid") in self.coordinator.game_icons:
return self.coordinator.game_icons[player["gameid"]]
# Reset game icons to have coordinator get id for new game
self.coordinator.game_icons = {}
if player.gameid is not None and player.gameid in self.coordinator.game_icons:
return self.coordinator.game_icons[player.gameid]
return None
@property
homeassistant/components/steam_online/strings.json
+11
View File
@@ -70,6 +70,17 @@
}
}
},
"exceptions": {
"auth_exception": {
"message": "Failed to connect to Steam due to an authentication error"
},
"request_exception": {
"message": "Failed to connect to Steam due to a request error"
},
"timeout_exception": {
"message": "Failed to connect to Steam due to a request timeout"
}
},
"options": {
"error": {
"unauthorized": "Friends list restricted: Please refer to the documentation on how to see all other friends"
tests/components/steam_online/snapshots/test_sensor.ambr
+1
View File
@@ -42,6 +42,7 @@
<EntityStateAttribute.ENTITY_PICTURE: 'entity_picture'>: 'https://avatars.steamstatic.com/fef49e7fa7e1997310d705b2a6158ff8dc1cdfeb_full.jpg',
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Steam testaccount1',
'game': 'The Witcher: Enhanced Edition',
'game_icon': 'https://steamcdn-a.akamaihd.net/steamcommunity/public/images/apps/20900/746d1cd48fb2e57d579b05b6e9eccba95859e549.jpg',
'game_id': '20900',
'game_image_header': 'https://steamcdn-a.akamaihd.net/steam/apps/20900/header.jpg',
'game_image_main': 'https://steamcdn-a.akamaihd.net/steam/apps/20900/capsule_616x353.jpg',
tests/components/steam_online/test_config_flow.py
+2 -2
View File
@@ -64,7 +64,7 @@ async def test_flow_user(
async def test_flow_user_errors(
hass: HomeAssistant,
mock_setup_entry: AsyncMock,
side_effect: Exception | dict[str, Any],
side_effect: type[Exception] | dict[str, Any],
error_msg: str,
steam_api: MagicMock,
) -> None:
@@ -157,7 +157,7 @@ async def test_flow_reauth_errors(
hass: HomeAssistant,
config_entry: MockConfigEntry,
steam_api: MagicMock,
side_effect: Exception | dict[str, Any],
side_effect: type[Exception] | dict[str, Any],
error_msg: str,
) -> None:
"""Test reauth step with errors."""
tests/components/steam_online/test_init.py
+18 -2
View File
@@ -35,30 +35,46 @@ async def test_setup(
assert not hass.data.get(DOMAIN)
@pytest.mark.parametrize(
"side_effect",
[
steam.api.HTTPError,
steam.api.HTTPTimeoutError,
],
)
async def test_setup_errors(
hass: HomeAssistant,
config_entry: MockConfigEntry,
steam_api: MagicMock,
side_effect: type[Exception],
) -> None:
"""Test setup errors."""
config_entry.add_to_hass(hass)
steam_api.return_value.GetPlayerSummaries.side_effect = steam.api.HTTPError
steam_api.return_value.GetPlayerSummaries.side_effect = side_effect
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()
assert config_entry.state is ConfigEntryState.SETUP_RETRY
@pytest.mark.parametrize(
"side_effect",
[
steam.api.HTTPError("Server connection failed: Forbidden (403)"),
steam.api.HTTPError("Server connection failed: Unauthorized (401)"),
],
)
async def test_setup_auth_failed(
hass: HomeAssistant,
config_entry: MockConfigEntry,
steam_api: MagicMock,
side_effect: type[Exception],
) -> None:
"""Test that it throws ConfigEntryAuthFailed when authentication fails."""
config_entry.add_to_hass(hass)
steam_api.return_value.GetPlayerSummaries.side_effect = steam.api.HTTPError("401")
steam_api.return_value.GetPlayerSummaries.side_effect = side_effect
await hass.config_entries.async_setup(config_entry.entry_id)
await hass.async_block_till_done()