diff --git a/light_api/light_api/client.py b/light_api/light_api/client.py index c8db179..9032200 100644 --- a/light_api/light_api/client.py +++ b/light_api/light_api/client.py @@ -6,7 +6,7 @@ import logging import os -from typing import TYPE_CHECKING, Any, Callable, NewType, final +from typing import TYPE_CHECKING, Any, Callable, Container, Iterable, Iterator, NewType, final from open_api_specification_client.api.default import get_api_playlists from open_api_specification_client.client import AuthenticatedClient @@ -21,6 +21,9 @@ from open_api_specification_client.models.get_api_devices_response_200 import ( GetApiDevicesResponse200, ) + from open_api_specification_client.models.get_api_devices_response_200_included_item import ( + GetApiDevicesResponse200IncludedItem, + ) KEYRING_SERVICE = "unofficial-light-api" KEYRING_USER = "session" @@ -121,6 +124,33 @@ def call_api(self, func: Callable[..., Any], **kwargs: Any) -> Any: resp = func(**kwargs) return resp + @staticmethod + def _ensure_ok( + resp: Any, + action: str, + ok_codes: Container[int] = (200,), + require_parsed: bool = False, + require_data: bool = False, + ) -> Any: + """Raise RuntimeError(f"{action}: {status}") unless resp is ok, else return resp.parsed. + + Args: + resp: The API response to parse + action: Description of the API request being validated. Used in error message. + ok_codes: Status codes that the caller considers a success + require_parsed: True if a non-None `resp.parsed` is required for success + require_data: True if a non-empty `resp.parsed.data` is required for success + (implies require_parsed) + + Returns: + The parsed data + """ + parsed_missing = (require_parsed or require_data) and resp.parsed is None + data_missing = require_data and not parsed_missing and not resp.parsed.data + if resp.status_code not in ok_codes or parsed_missing or data_missing: + raise RuntimeError(f"{action}: {resp.status_code}") + return resp.parsed + def __enter__(self) -> Light: """Sets up API session.""" log.info("Authenticating") @@ -224,9 +254,8 @@ def _fetch_playlist_id(self) -> None: client=self._api_client, device_tool_id=music_id, ) - if resp.status_code != 200 or not resp.parsed or not resp.parsed.data: - raise RuntimeError(f"Could not fetch playlists: {resp.status_code}") - self._playlist_id = resp.parsed.data[0].id + parsed = self._ensure_ok(resp, "Could not fetch playlists", require_data=True) + self._playlist_id = parsed.data[0].id def _fetch_device_tool_ids(self) -> None: """Populate _device_tool_ids for all installed tools. @@ -249,31 +278,19 @@ def _fetch_device_tool_ids(self) -> None: ) devices_resp = get_api_devices.sync_detailed(client=self._api_client) - if ( - devices_resp.status_code != 200 - or not devices_resp.parsed - or not devices_resp.parsed.data - ): - raise RuntimeError(f"Could not fetch devices: {devices_resp.status_code}") - device_id = self._select_device_id(devices_resp.parsed) + devices = self._ensure_ok(devices_resp, "Could not fetch devices", require_data=True) + device_id = self._select_device_id(devices) tools_resp = get_api_tools.sync_detailed( client=self._api_client, device_id=device_id ) - if tools_resp.status_code != 200 or not tools_resp.parsed: - raise RuntimeError(f"Could not fetch tools: {tools_resp.status_code}") + tools = self._ensure_ok(tools_resp, "Could not fetch tools", require_parsed=True) tool_ns: dict[str, str] = { - t.id: t.attributes.namespace.lower() for t in tools_resp.parsed.data + t.id: t.attributes.namespace.lower() for t in tools.data } - for item in devices_resp.parsed.included: - if isinstance(item.relationships, Unset) or isinstance( - item.relationships.tool, Unset - ): - continue - if item.relationships.device.data.id != device_id: - continue + for item in self._device_tool_items(devices.included, device_id): ns = tool_ns.get(item.relationships.tool.data.id, "") if "note" in ns: self._device_tool_ids["notes"] = item.id @@ -282,6 +299,32 @@ def _fetch_device_tool_ids(self) -> None: elif "music" in ns or "playlist" in ns: self._device_tool_ids["music"] = item.id + @staticmethod + def _device_tool_items( + included: Iterable[GetApiDevicesResponse200IncludedItem], device_id: DeviceId + ) -> Iterator[GetApiDevicesResponse200IncludedItem]: + """Yield the device_tool items in `included` belonging to `device_id`.""" + for item in included: + if isinstance(item.relationships, Unset) or isinstance( + item.relationships.tool, Unset + ): + continue + if item.relationships.device.data.id != device_id: + continue + yield item + + @staticmethod + def _device_phone_numbers( + included: Iterable[GetApiDevicesResponse200IncludedItem], + ) -> Iterator[tuple[DeviceId, PhoneNumber]]: + """Yield (device_id, phone_number) pairs from the sims records in `included`.""" + for item in included: + if item.type_ != "sims" or isinstance(item.attributes.phone_number, Unset): + continue + yield DeviceId(item.relationships.device.data.id), PhoneNumber( + item.attributes.phone_number + ) + def _select_device_id(self, devices: GetApiDevicesResponse200) -> DeviceId: """Select the correct device id out of /api/devices data. @@ -309,17 +352,9 @@ def _select_device_id(self, devices: GetApiDevicesResponse200) -> DeviceId: target = self._phone_digits(self.phone) seen: list[str] = [] - for item in devices.included: - if item.type_ != "sims" or isinstance( - item.attributes.phone_number, Unset - ): - continue - - device_id = DeviceId(item.relationships.device.data.id) - number = PhoneNumber(item.attributes.phone_number) + for device_id, number in self._device_phone_numbers(devices.included): if self._phone_digits(number) == target: return device_id - seen.append(f"{device_id} ({number})") raise RuntimeError( diff --git a/light_api/light_api/devices.py b/light_api/light_api/devices.py index 54dc5bf..b0f18e5 100644 --- a/light_api/light_api/devices.py +++ b/light_api/light_api/devices.py @@ -3,8 +3,6 @@ from dataclasses import dataclass from typing import TYPE_CHECKING -from open_api_specification_client.types import Unset - if TYPE_CHECKING: from light_api.client import Light @@ -28,16 +26,9 @@ def list_devices(self) -> list[LightDevice]: resp = self._l.call_api( get_api_devices.sync_detailed, client=self._l._api_client ) - if resp.status_code != 200 or not resp.parsed or not resp.parsed.data: - raise RuntimeError(f"Could not fetch devices: {resp.status_code}") - - phone_by_device: dict[str, str] = {} - for item in resp.parsed.included: - if item.type_ != "sims" or isinstance(item.attributes.phone_number, Unset): - continue - phone_by_device[item.relationships.device.data.id] = ( - item.attributes.phone_number - ) + devices = self._l._ensure_ok(resp, "Could not fetch devices", require_data=True) + + phone_by_device = dict(self._l._device_phone_numbers(devices.included)) return [ LightDevice( @@ -46,5 +37,5 @@ def list_devices(self) -> list[LightDevice]: serial_number=d.attributes.serial_number, sku=d.attributes.sku, ) - for d in resp.parsed.data + for d in devices.data ] diff --git a/light_api/light_api/music.py b/light_api/light_api/music.py index 46e658f..22e2fa1 100644 --- a/light_api/light_api/music.py +++ b/light_api/light_api/music.py @@ -105,10 +105,9 @@ def get_sort_mode(self) -> SortMode: client=self._l._api_client, device_tool_id=self._l._device_tool_ids["music"], ) - if resp.status_code != 200 or resp.parsed is None: - raise RuntimeError(f"Get sort mode: {resp.status_code}") + parsed = self._l._ensure_ok(resp, "Get sort mode", require_parsed=True) - playlist = resp.parsed.data[0] + playlist = parsed.data[0] return SortMode(playlist.attributes.sort_mode) def set_sort_mode(self, sort_mode: SortMode): @@ -137,8 +136,7 @@ def set_sort_mode(self, sort_mode: SortMode): ), ) - if not (200 <= resp.status_code < 300): - raise RuntimeError(f"Set sort mode: {resp.status_code}") + self._l._ensure_ok(resp, "Set sort mode", ok_codes=range(200, 300)) log.info(f"Sort mode set") elif sort_mode in (SortMode.TITLE_ASC, SortMode.TITLE_DESC): @@ -158,10 +156,7 @@ def get_tracks(self) -> list[LightTrack]: playlist_ids=self._l._playlist_id, device_tool_id=self._l._device_tool_ids["music"], ) - if resp.status_code != 200 or resp.parsed is None: - raise RuntimeError(f"Get tracks: {resp.status_code}") - - body = resp.parsed + body = self._l._ensure_ok(resp, "Get tracks", require_parsed=True) if not body.data: return [] @@ -204,8 +199,7 @@ def delete_all_tracks(self) -> None: device_tool_id=self._l._device_tool_ids["music"] ), ) - if not (200 <= resp.status_code < 300): - raise RuntimeError(f"Failed to delete all tracks: {resp.status_code}") + self._l._ensure_ok(resp, "Failed to delete all tracks", ok_codes=range(200, 300)) log.info("All tracks deleted") def delete_tracks_predicate(self, predicate: Callable[[LightTrack], bool]) -> None: @@ -336,14 +330,16 @@ def upload_tracks( ) ), ) - if create_resp.status_code not in (200, 201) or create_resp.parsed is None: - raise RuntimeError( - f"Create audio record for {os.path.basename(upload_path)}: {create_resp.status_code}" - ) + created = self._l._ensure_ok( + create_resp, + f"Create audio record for {os.path.basename(upload_path)}", + ok_codes=(200, 201), + require_parsed=True, + ) presigned_url = next( item.attributes.presigned_url - for item in create_resp.parsed.included + for item in created.included if item.type_ == "files" ) @@ -431,8 +427,7 @@ def update_track_metadata( ) ), ) - if not (200 <= resp.status_code < 300): - raise RuntimeError(f"update metadata: {resp.status_code}") + self._l._ensure_ok(resp, "update metadata", ok_codes=range(200, 300)) log.info("Metadata updated") @@ -488,10 +483,9 @@ def reorder_subset(self, ordered_item_ids: list[str]) -> None: ) ), ) - if not (200 <= resp.status_code < 300): - raise RuntimeError( - f"reorder_subset position {new_position}: {resp.status_code}" - ) + self._l._ensure_ok( + resp, f"reorder_subset position {new_position}", ok_codes=range(200, 300) + ) def _apply_sort_positions(self, sorted_tracks: list[LightTrack], original_tracks: list[LightTrack]) -> None: """PATCH playlist item positions to match the given sort order.""" @@ -513,8 +507,9 @@ def _apply_sort_positions(self, sorted_tracks: list[LightTrack], original_tracks ) ), ) - if not (200 <= resp.status_code < 300): - raise RuntimeError(f"Apply sort position {new_position}: {resp.status_code}") + self._l._ensure_ok( + resp, f"Apply sort position {new_position}", ok_codes=range(200, 300) + ) def _sort_by_title(self, descending: bool = False) -> None: """Sort tracks on device by title. diff --git a/light_api/light_api/notes.py b/light_api/light_api/notes.py index 7b788b8..2a1b50e 100644 --- a/light_api/light_api/notes.py +++ b/light_api/light_api/notes.py @@ -67,8 +67,7 @@ def get_note_content(self, note: LightNote) -> bytes: note_id=note.id, client=self._l._api_client, ) - if resp.status_code != 200 or resp.parsed is None: - raise RuntimeError(f"Presigned get URL for {note.id}: {resp.status_code}") + self._l._ensure_ok(resp, f"Presigned get URL for {note.id}", require_parsed=True) content_resp = httpx.get(resp.parsed.presigned_get_url, timeout=30) if not content_resp.is_success: @@ -81,10 +80,7 @@ def get_notes(self) -> list["LightNote"]: client=self._l._api_client, device_tool_id=self._l._device_tool_ids["notes"], ) - if resp.status_code != 200 or resp.parsed is None: - raise RuntimeError(f"List notes: {resp.status_code}") - - body = resp.parsed + body = self._l._ensure_ok(resp, "List notes", require_parsed=True) return [_make_light_note(data) for data in body.data] @@ -95,8 +91,7 @@ def get_note_metadata(self, note_id: str) -> "LightNote": client=self._l._api_client, device_tool_id=self._l._device_tool_ids["notes"], ) - if resp.status_code != 200 or resp.parsed is None: - raise RuntimeError(f"Fetching note {note_id}: {resp.status_code}") + self._l._ensure_ok(resp, f"Fetching note {note_id}", require_parsed=True) return _make_light_note(resp.parsed.data) @@ -147,10 +142,11 @@ def create_text_note( ) ), ) - if resp.status_code not in (200, 201) or resp.parsed is None: - raise RuntimeError(f"Creating note: {resp.status_code}") + parsed = self._l._ensure_ok( + resp, "Creating note", ok_codes=(200, 201), require_parsed=True + ) - presigned_url = resp.parsed.included[0].attributes.presigned_url + presigned_url = parsed.included[0].attributes.presigned_url if content_is_path: with open(content) as f: @@ -163,7 +159,7 @@ def create_text_note( if not put_resp.is_success: raise RuntimeError(f"Upload note content: {put_resp.status_code}") - note = _make_light_note(resp.parsed.data) + note = _make_light_note(parsed.data) log.info(f"Note {note.id} created") return note @@ -178,8 +174,7 @@ def update_note_content(self, note: LightNote, content: bytes) -> None: client=self._l._api_client, note_id=note.id, ) - if resp.status_code != 200 or resp.parsed is None: - raise RuntimeError(f"Presigned put URL for {note.id}: {resp.status_code}") + self._l._ensure_ok(resp, f"Presigned put URL for {note.id}", require_parsed=True) put_resp = httpx.put(resp.parsed.presigned_put_url, content=content, timeout=30) if not put_resp.is_success: raise RuntimeError(f"Upload note content: {put_resp.status_code}") @@ -211,8 +206,7 @@ def update_note_title(self, note: LightNote, title: str) -> None: ) ), ) - if resp.status_code not in (200, 204): - raise RuntimeError(f"Update note title: {resp.status_code}") + self._l._ensure_ok(resp, "Update note title", ok_codes=(200, 204)) note.title = title log.info(f"Note {note.id} title updated to {title!r}") @@ -223,6 +217,5 @@ def delete_note(self, note_id: str) -> None: client=self._l._api_client, note_id=note_id, ) - if resp.status_code not in (200, 204): - raise RuntimeError(f"Delete note: {resp.status_code}") + self._l._ensure_ok(resp, "Delete note", ok_codes=(200, 204)) log.info(f"Note {note_id} deleted") diff --git a/light_api/light_api/podcast.py b/light_api/light_api/podcast.py index 78ecf47..4482dfb 100644 --- a/light_api/light_api/podcast.py +++ b/light_api/light_api/podcast.py @@ -65,10 +65,7 @@ def get_podcasts(self) -> list[LightPodcast]: client=self._l._api_client, device_tool_id=self._l._device_tool_ids["podcast"], ) - if resp.status_code != 200 or resp.parsed is None: - raise RuntimeError(f"Get podcasts: {resp.status_code}") - - body = resp.parsed + body = self._l._ensure_ok(resp, "Get podcasts", require_parsed=True) included_by_id = { item.id: item.attributes for item in body.included @@ -109,8 +106,7 @@ def delete_podcast_by_title(self, title: str) -> None: followed_podcast_id=p.followed_podcast_id, client=self._l._api_client, ) - if not (200 <= resp.status_code < 300): - raise RuntimeError(f"Delete podcast: {resp.status_code}") + self._l._ensure_ok(resp, "Delete podcast", ok_codes=range(200, 300)) def add_podcast(self, rss_feed_url: str) -> LightPodcast: """Add a podcast to the device by RSS feed URL. @@ -134,11 +130,12 @@ def add_podcast(self, rss_feed_url: str) -> LightPodcast: ) ), ) - if create_resp.status_code not in (200, 201) or create_resp.parsed is None: - raise RuntimeError(f"Create podcast: {create_resp.status_code}") + created = self._l._ensure_ok( + create_resp, "Create podcast", ok_codes=(200, 201), require_parsed=True + ) - podcast_id = create_resp.parsed.data.id - attrs = create_resp.parsed.data.attributes + podcast_id = created.data.id + attrs = created.data.attributes follow_resp = post_api_followed_podcasts.sync_detailed( client=self._l._api_client, @@ -162,10 +159,11 @@ def add_podcast(self, rss_feed_url: str) -> LightPodcast: ) ), ) - if follow_resp.status_code not in (200, 201) or follow_resp.parsed is None: - raise RuntimeError(f"Follow podcast: {follow_resp.status_code}") + followed = self._l._ensure_ok( + follow_resp, "Follow podcast", ok_codes=(200, 201), require_parsed=True + ) - followed_podcast_id = follow_resp.parsed.data.id + followed_podcast_id = followed.data.id return LightPodcast( podcast_id=podcast_id, diff --git a/light_api/light_api/tools.py b/light_api/light_api/tools.py index f15ad74..403312f 100644 --- a/light_api/light_api/tools.py +++ b/light_api/light_api/tools.py @@ -45,41 +45,26 @@ def get_tools(self) -> list[LightTool]: get_api_devices, get_api_tools, ) - from open_api_specification_client.types import Unset devices_resp = self._l.call_api( get_api_devices.sync_detailed, client=self._l._api_client ) + devices = self._l._ensure_ok(devices_resp, "Could not fetch devices", require_data=True) - if ( - devices_resp.status_code != 200 - or not devices_resp.parsed - or not devices_resp.parsed.data - ): - raise RuntimeError(f"Could not fetch devices: {devices_resp.status_code}") - - device_id = self._l._select_device_id(devices_resp.parsed) + device_id = self._l._select_device_id(devices) tools_resp = get_api_tools.sync_detailed( client=self._l._api_client, device_id=device_id ) - - if tools_resp.status_code != 200 or not tools_resp.parsed: - raise RuntimeError(f"Could not fetch tools: {tools_resp.status_code}") + tools = self._l._ensure_ok(tools_resp, "Could not fetch tools", require_parsed=True) tool_info = { t.id: (t.attributes.namespace, t.attributes.component, t.attributes.title) - for t in tools_resp.parsed.data + for t in tools.data } results = [] - for item in devices_resp.parsed.included: - if isinstance(item.relationships, Unset) or isinstance( - item.relationships.tool, Unset - ): - continue - if item.relationships.device.data.id != device_id: - continue + for item in self._l._device_tool_items(devices.included, device_id): global_tool_id = item.relationships.tool.data.id ns, comp, title = tool_info.get(global_tool_id, ("", "", "")) results.append( @@ -97,20 +82,18 @@ def get_tools(self) -> list[LightTool]: def _get_device_id(self) -> str: from open_api_specification_client.api.default import get_api_devices resp = self._l.call_api(get_api_devices.sync_detailed, client=self._l._api_client) - if resp.status_code != 200 or not resp.parsed or not resp.parsed.data: - raise RuntimeError(f"Could not fetch devices: {resp.status_code}") - return self._l._select_device_id(resp.parsed) + devices = self._l._ensure_ok(resp, "Could not fetch devices", require_data=True) + return self._l._select_device_id(devices) def _resolve_global_tool_id(self, name: str) -> tuple[str, str]: """Return (global_tool_id, title) for a tool matching name (case-insensitive).""" from open_api_specification_client.api.default import get_api_tools device_id = self._get_device_id() resp = get_api_tools.sync_detailed(client=self._l._api_client, device_id=device_id) - if resp.status_code != 200 or not resp.parsed: - raise RuntimeError(f"Could not fetch tools: {resp.status_code}") + tools = self._l._ensure_ok(resp, "Could not fetch tools", require_parsed=True) needle = name.lower() matches = [ - t for t in resp.parsed.data + t for t in tools.data if needle in t.attributes.title.lower() or needle in t.attributes.namespace.lower() ] if not matches: @@ -146,11 +129,12 @@ def add_tool(self, name: ToolName | str) -> LightTool: ) ), ) - if resp.status_code not in (200, 201) or resp.parsed is None: - raise RuntimeError(f"Install tool: {resp.status_code}") + parsed = self._l._ensure_ok( + resp, "Install tool", ok_codes=(200, 201), require_parsed=True + ) return LightTool( - device_tool_id=resp.parsed.data.id, + device_tool_id=parsed.data.id, global_tool_id=global_tool_id, namespace="", component="", @@ -175,5 +159,4 @@ def remove_tool(self, name: ToolName | str) -> None: client=self._l._api_client, device_tool_id=matches[0].device_tool_id, ) - if not (200 <= resp.status_code < 300): - raise RuntimeError(f"Remove tool: {resp.status_code}") + self._l._ensure_ok(resp, "Remove tool", ok_codes=range(200, 300)) diff --git a/tests/test_api.py b/tests/test_api.py index 78f1a55..840bc1c 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -4,6 +4,7 @@ import pytest import respx import httpx +from types import SimpleNamespace from unittest.mock import patch from light_api.client import Light @@ -30,6 +31,61 @@ def make_light(phone: str | None = None, device_id: str | None = None) -> Light: return light +def fake_resp(status_code: int, parsed=None): + return SimpleNamespace(status_code=status_code, parsed=parsed) + + +class TestEnsureOk: + def test_returns_parsed_on_success(self): + parsed = SimpleNamespace(data=["x"]) + result = Light._ensure_ok(fake_resp(200, parsed), "Do thing") + assert result is parsed + + def test_raises_on_unexpected_status(self): + with pytest.raises(RuntimeError, match="Do thing: 404"): + Light._ensure_ok(fake_resp(404), "Do thing") + + def test_accepts_alternate_ok_codes(self): + parsed = SimpleNamespace(data=["x"]) + result = Light._ensure_ok(fake_resp(201, parsed), "Create thing", ok_codes=(200, 201)) + assert result is parsed + + def test_accepts_status_only_range(self): + result = Light._ensure_ok(fake_resp(204, None), "Delete thing", ok_codes=range(200, 300)) + assert result is None + + def test_require_data_raises_on_empty_data(self): + parsed = SimpleNamespace(data=[]) + with pytest.raises(RuntimeError, match="Do thing: 200"): + Light._ensure_ok(fake_resp(200, parsed), "Do thing", require_data=True) + + def test_require_data_raises_on_none_parsed(self): + """require_data=True alone (no require_parsed) still catches parsed=None - it's tiered.""" + with pytest.raises(RuntimeError, match="Do thing: 200"): + Light._ensure_ok(fake_resp(200, None), "Do thing", require_data=True) + + def test_require_data_succeeds_with_data(self): + parsed = SimpleNamespace(data=["x"]) + result = Light._ensure_ok(fake_resp(200, parsed), "Do thing", require_data=True) + assert result is parsed + + def test_require_parsed_raises_on_none_parsed(self): + with pytest.raises(RuntimeError, match="Do thing: 200"): + Light._ensure_ok(fake_resp(200, None), "Do thing", require_parsed=True) + + def test_require_parsed_allows_empty_data(self): + """require_parsed=True does NOT imply require_data - empty .data is fine.""" + parsed = SimpleNamespace(data=[]) + result = Light._ensure_ok(fake_resp(200, parsed), "Do thing", require_parsed=True) + assert result is parsed + + def test_default_allows_none_parsed(self): + """Neither flag set - only status is checked, matching endpoints like + delete/update that don't touch resp.parsed afterward.""" + result = Light._ensure_ok(fake_resp(204, None), "Do thing", ok_codes=(200, 204)) + assert result is None + + class TestFetchDeviceToolIds: @respx.mock def test_populates_all_tool_ids(self, f_devices, f_tools):