Refactor/dedupe api response checks

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(

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
         ]

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.

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")