pyalexatodo.api

  1from http import HTTPMethod
  2from typing import TYPE_CHECKING
  3
  4from aioamazondevices.api import AmazonEchoApi
  5
  6from pyalexatodo.exceptions import ItemNotFoundException
  7from pyalexatodo.models.list_info import ListInfo
  8from pyalexatodo.models.list_response import ListResponse
  9from pyalexatodo.models.list_item_status import ListItemStatus
 10from pyalexatodo.models.list_items_response import ListItem, ListItemsResponse
 11
 12if TYPE_CHECKING:
 13    from aiohttp import ClientResponse
 14
 15
 16class AlexaToDoAPI:
 17    """A client for interacting with Amazon Alexa shopping lists.
 18
 19    This class provides methods to manage Alexa shopping lists, including:
 20    - Fetching lists and list items
 21    - Adding, removing, and updating items
 22    - Managing item status (checked/unchecked)
 23
 24    All methods are asynchronous and require an active AmazonEchoApi login session.
 25    """
 26    def __init__(self, alexa_echo_api: AmazonEchoApi, base_url: str | None = None):
 27        """Initialize the Alexa List API client.
 28
 29        Args:
 30            alexa_echo_api: An authenticated AmazonEchoApi instance.
 31            base_url: Base URL for API requests (for testing). If None, uses Amazon's URL.
 32        """
 33        self.alexa_echo_api = alexa_echo_api
 34        """AmazonEchoAPI instance used for making authenticated requests to the Alexa API."""
 35
 36        self._domain_extension = alexa_echo_api.domain
 37        self._base_url = base_url or f"https://www.amazon.{self._domain_extension}"
 38
 39    async def _http_request(
 40        self, method: HTTPMethod, url: str, data: dict
 41    ) -> "ClientResponse":
 42        """
 43        Make an HTTP request to the Alexa API.
 44
 45        Args:
 46            method: The HTTP method to use (GET, POST, PUT, DELETE, etc.).
 47            url: The URL endpoint to request.
 48            data: The request data to be sent as JSON.
 49
 50        Returns:
 51            The response object from the HTTP request.
 52        """
 53        _, response = await self.alexa_echo_api._http_wrapper.session_request(
 54            method=method, url=url, input_data=data, json_data=True,
 55        )
 56
 57        return response
 58
 59    async def get_lists(self) -> ListInfo[ListInfo]:
 60        """Fetch all available Alexa shopping lists.
 61
 62        Returns:
 63            A list of shopping list information objects.
 64
 65        Raises:
 66            Exception: If the API request fails.
 67        """
 68        result = await self._http_request(
 69            HTTPMethod.POST,
 70            f"{self._base_url}/alexashoppinglists/api/v2/lists/fetch",
 71            {},
 72        )
 73
 74        if not result or result.status != 200:
 75            raise Exception("Failed to fetch lists")
 76
 77        result_json = await result.json()
 78        list_infos = ListResponse(**result_json)
 79
 80        return list_infos.listInfoList
 81
 82    async def get_list_items(self, list_id: str) -> ListInfo[ListItem]:
 83        """Fetch all items from a specified Alexa shopping list.
 84
 85        Args:
 86            list_id: The ID of the list to fetch items from.
 87
 88        Returns:
 89            A list of shopping list items.
 90
 91        Raises:
 92            Exception: If the API request fails.
 93        """
 94        result = await self._http_request(
 95            HTTPMethod.POST,
 96            f"{self._base_url}/alexashoppinglists/api/v2/lists/{list_id}/items/fetch?limit=100",
 97            {},
 98        )
 99
100        if not result or result.status != 200:
101            raise Exception(f"Failed to fetch list items for list: {list_id}")
102
103        result_json = await result.json()
104        list_items = ListItemsResponse(**result_json)
105
106        return list_items.itemInfoList
107
108    async def set_item_checked_status(
109        self, list_id: str, item_id: str, checked: bool, version: int
110    ):
111        """Update the checked status of an item in a shopping list.
112
113        Args:
114            list_id: The ID of the list containing the item.
115            item_id: The ID of the item to update.
116            checked: True to mark as complete, False to mark as active.
117            version: The current version of the item.
118                     The value is included in the get_list_items response and is required by the Amazon API.
119
120        Raises:
121            Exception: If the API request fails.
122        """
123        result = await self._http_request(
124            HTTPMethod.PUT,
125            f"{self._base_url}/alexashoppinglists/api/v2/lists/{list_id}/items/{item_id}?version={version}",
126            {
127                "itemAttributesToUpdate": [
128                    {
129                        "type": "itemStatus",
130                        "value": ListItemStatus.COMPLETE.value
131                        if checked
132                        else ListItemStatus.ACTIVE.value,
133                    }
134                ],
135                "itemAttributesToRemove": [],
136            },
137        )
138
139        if not result or result.status != 200:
140            raise Exception(f"Failed to toggle item: {item_id}")
141
142    async def add_item(self, list_id: str, name: str):
143        """Add a new item to a shopping list.
144
145        Args:
146            list_id: The ID of the list to add the item to.
147            name: The name of the item to add.
148
149        Raises:
150            Exception: If the API request fails.
151        """
152        result = await self._http_request(
153            HTTPMethod.POST,
154            f"{self._base_url}/alexashoppinglists/api/v2/lists/{list_id}/items",
155            {
156                "items": [
157                    {
158                        "itemType": "KEYWORD",
159                        "itemName": name,
160                    }
161                ]
162            },
163        )
164
165        if not result or result.status != 200:
166            raise Exception(f"Failed to add item: {name}")
167
168    async def delete_item(self, list_id: str, item_id: str, version: int):
169        """Delete an item from a shopping list.
170
171        Args:
172            list_id: The ID of the list containing the item.
173            item_id: The ID of the item to delete.
174            version: The current version of the item.
175                     The value is included in the get_list_items response and is required by the Amazon API.
176
177        Raises:
178            Exception: If the API request fails.
179        """
180        result = await self._http_request(
181            HTTPMethod.DELETE,
182            f"{self._base_url}/alexashoppinglists/api/v2/lists/{list_id}/items/{item_id}?version={version}",
183            {},
184        )
185
186        if not result or result.status != 200:
187            raise Exception(f"Failed to delete item: {item_id}")
188
189    async def rename_item(
190        self, list_id: str, item_id: str, new_name: str, version: int
191    ):
192        """Rename an item in a shopping list.
193
194        Args:
195            list_id: The ID of the list containing the item.
196            item_id: The ID of the item to rename.
197            new_name: The new name for the item.
198            version: The current version of the item.
199                     The value is included in the get_list_items response and is required by the Amazon API.
200
201        Raises:
202            Exception: If the API request fails.
203        """
204        result = await self._http_request(
205            HTTPMethod.PUT,
206            f"{self._base_url}/alexashoppinglists/api/v2/lists/{list_id}/items/{item_id}?version={version}",
207            {
208                "itemAttributesToUpdate": [{"type": "itemName", "value": new_name}],
209                "itemAttributesToRemove": [],
210            },
211        )
212
213        if not result or result.status != 200:
214            raise Exception(f"Failed to rename item: {item_id}")
215
216    async def get_item_by_name(self, list_id: str, name: str) -> ListItem:
217        """Find an item in a shopping list by its name.
218
219        The search is case-insensitive. If multiple items have the same name,
220        the first one found will be returned.
221
222        Args:
223            list_id: The ID of the list to search in.
224            name: The name of the item to find.
225
226        Returns:
227            The found item.
228
229        Raises:
230            ItemNotFoundException: If no item with the given name is found.
231        """
232        list_items = await self.get_list_items(list_id)
233        for list_item in list_items:
234            if list_item.name.casefold() == name.casefold():
235                return list_item
236        raise ItemNotFoundException