Add method to retrieve list activity

bring_api/__init__.py

@@ -12,6 +12,7 @@
     BringUserUnknownException,
 )
 from .types import (
+    BringActivityResponse,
     BringAuthResponse,
     BringAuthTokenResponse,
     BringItem,
@@ -30,6 +31,7 @@
 
 __all__ = [
     "Bring",
+    "BringActivityResponse",
     "BringAuthException",
     "BringAuthResponse",
     "BringAuthTokenResponse",

bring_api/bring.py

@@ -32,6 +32,7 @@
     BringUserUnknownException,
 )
 from .types import (
+    BringActivityResponse,
     BringAuthResponse,
     BringAuthTokenResponse,
     BringItem,
@@ -1477,3 +1478,62 @@ async def set_list_article_language(
             raise BringRequestException(
                 "Set list article language failed due to request exception."
             ) from e
+
+    async def get_activity(self, list_uuid: UUID) -> BringActivityResponse:
+        """Get activity for given list."""
+        try:
+            url = self.url / "v2/bringlists" / str(list_uuid) / "activity"
+            async with self._session.get(url, headers=self.headers) as r:
+                _LOGGER.debug(
+                    "Response from %s [%s]: %s", url, r.status, await r.text()
+                )
+
+                if r.status == HTTPStatus.UNAUTHORIZED:
+                    try:
+                        errmsg = await r.json()
+                    except (JSONDecodeError, aiohttp.ClientError):
+                        _LOGGER.debug(
+                            "Exception: Cannot parse request response:\n %s",
+                            traceback.format_exc(),
+                        )
+                    else:
+                        _LOGGER.debug(
+                            "Exception: Cannot get list activity: %s", errmsg["message"]
+                        )
+                    raise BringAuthException(
+                        "Loading list activity failed due to authorization failure, "
+                        "the authorization token is invalid or expired."
+                    )
+
+                r.raise_for_status()
+
+                try:
+                    return BringActivityResponse.from_json(await r.text())
+                except (JSONDecodeError, KeyError) as e:
+                    _LOGGER.debug(
+                        "Exception: Cannot get activity for list %s:\n%s",
+                        list_uuid,
+                        traceback.format_exc(),
+                    )
+                    raise BringParseException(
+                        "Loading list activity failed during parsing of request response."
+                    ) from e
+
+        except TimeoutError as e:
+            _LOGGER.debug(
+                "Exception: Cannot get activity for list %s:\n%s",
+                list_uuid,
+                traceback.format_exc(),
+            )
+            raise BringRequestException(
+                "Loading list activity failed due to connection timeout."
+            ) from e
+        except aiohttp.ClientError as e:
+            _LOGGER.debug(
+                "Exception: Cannot get activity for list %s:\n%s",
+                list_uuid,
+                traceback.format_exc(),
+            )
+            raise BringRequestException(
+                "Loading list activity failed due to request exception."
+            ) from e

bring_api/types.py

@@ -1,6 +1,7 @@
 """Bring API types."""
 
 from dataclasses import dataclass, field
+from datetime import datetime
 from enum import Enum, StrEnum
 from typing import Literal, NotRequired, TypedDict
 from uuid import UUID
@@ -231,3 +232,40 @@ class BringAuthTokenResponse(DataClassORJSONMixin):
     refresh_token: str
     token_type: str
     expires_in: int
+
+
+class ActivityType(StrEnum):
+    """Activity type."""
+
+    LIST_ITEMS_CHANGED = "LIST_ITEMS_CHANGED"
+    LIST_ITEMS_ADDED = "LIST_ITEMS_ADDED"
+    LIST_ITEMS_REMOVED = "LIST_ITEMS_REMOVED"
+
+
+@dataclass
+class ActivityContent:
+    """An activity content entry."""
+
+    uuid: UUID
+    sessionDate: datetime
+    publicUserUuid: UUID
+    items: list[BringPurchase] = field(default_factory=list)
+    purchase: list[BringPurchase] = field(default_factory=list)
+    recently: list[BringPurchase] = field(default_factory=list)
+
+
+@dataclass
+class Activity:
+    """An activity entry."""
+
+    type: ActivityType
+    content: ActivityContent
+
+
+@dataclass(kw_only=True)
+class BringActivityResponse(DataClassORJSONMixin):
+    """A list activity."""
+
+    timeline: list[Activity] = field(default_factory=list)
+    timestamp: datetime
+    totalEvents: int

tests/__snapshots__/test_bring.ambr

@@ -1,9 +1,12 @@
 # serializer version: 1
+# name: TestGetActivity.test_get_activity
+  BringActivityResponse(timeline=[Activity(type=<ActivityType.LIST_ITEMS_CHANGED: 'LIST_ITEMS_CHANGED'>, content=ActivityContent(uuid=UUID('673594a9-f92d-4cb6-adf1-d2f7a83207a4'), sessionDate=datetime.datetime(2025, 1, 1, 3, 9, 33, 36000, tzinfo=datetime.timezone.utc), publicUserUuid=UUID('98615d7e-0a7d-4a7e-8f73-a9cbb9f1bc32'), items=[], purchase=[BringPurchase(uuid=UUID('658a3770-1a03-4ee0-94a6-10362a642377'), itemId='Gurke', specification='', attributes=[])], recently=[BringPurchase(uuid=UUID('1ed22d3d-f19b-4530-a518-19872da3fd3e'), itemId='Milch', specification='', attributes=[])])), Activity(type=<ActivityType.LIST_ITEMS_ADDED: 'LIST_ITEMS_ADDED'>, content=ActivityContent(uuid=UUID('9a16635c-dea2-4e00-904a-c5034f9cfecf'), sessionDate=datetime.datetime(2025, 1, 1, 2, 54, 57, 656000, tzinfo=datetime.timezone.utc), publicUserUuid=UUID('6743a171-247d-46d0-bc06-baf31194f949'), items=[BringPurchase(uuid=UUID('66a633a2-ae09-47bf-8845-3c0198480544'), itemId='Joghurt', specification='', attributes=[])], purchase=[], recently=[])), Activity(type=<ActivityType.LIST_ITEMS_REMOVED: 'LIST_ITEMS_REMOVED'>, content=ActivityContent(uuid=UUID('303dedf6-d4b2-4d25-a8cd-1c7967b84fcb'), sessionDate=datetime.datetime(2025, 1, 1, 3, 9, 12, 380000, tzinfo=datetime.timezone.utc), publicUserUuid=UUID('6d79d10b-70b2-443f-9f7e-0b02e670c402'), items=[BringPurchase(uuid=UUID('2ba8ddb6-01c6-4b0b-a89d-f3da6b291528'), itemId='Tofu', specification='', attributes=[])], purchase=[], recently=[]))], timestamp=datetime.datetime(2025, 1, 1, 3, 9, 33, 36000, tzinfo=datetime.timezone.utc), totalEvents=2)
+# ---
 # name: TestGetAllItemDetails.test_get_all_item_details
   BringListItemsDetailsResponse(items=[BringListItemDetails(uuid=UUID('bfb5634c-d219-4d66-b68e-1388e54f0bb0'), itemId='Milchreis', listUuid=UUID('00000000-0000-0000-0000-000000000000'), userIconItemId='Reis', userSectionId='Getreideprodukte', assignedTo='', imageUrl=''), BringListItemDetails(uuid=UUID('0056b23c-7fc3-44da-8c34-426f8b632220'), itemId='Zitronensaft', listUuid=UUID('00000000-0000-0000-0000-000000000000'), userIconItemId='Zitrone', userSectionId='Zutaten & Gewürze', assignedTo='', imageUrl='')])
 # ---
 # name: TestGetAllUserSettings.test_get_all_user_settings
-  BringUserSettingsResponse(usersettings=[BringUserSettingsEntry(key='autoPush', value='ON'), BringUserSettingsEntry(key='purchaseStyle', value='grouped'), BringUserSettingsEntry(key='premiumHideSponsoredCategories', value='OFF'), BringUserSettingsEntry(key='premiumHideInspirationsBadge', value='OFF'), BringUserSettingsEntry(key='premiumHideOffersBadge', value='OFF'), BringUserSettingsEntry(key='premiumHideOffersOnMain', value='OFF'), BringUserSettingsEntry(key='defaultListUUID', value='xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxx'), BringUserSettingsEntry(key='discountActivatorOnMainEnabled', value='OFF'), BringUserSettingsEntry(key='onboardClient', value='android')], userlistsettings=[BringUserListSettingEntry(listUuid='00000000-00000000-00000000-00000000', usersettings=[BringUserSettingsEntry(key='listSectionOrder', value='["Früchte & Gemüse","Brot & Gebäck","Milch & Käse","Fleisch & Fisch","Zutaten & Gewürze","Fertig- & Tiefkühlprodukte","Getreideprodukte","Snacks & Süsswaren","Getränke & Tabak","Haushalt & Gesundheit","Pflege & Gesundheit","Tierbedarf","Baumarkt & Garten","Eigene Artikel"]'), BringUserSettingsEntry(key='listArticleLanguage', value='de-DE')])])
+  BringUserSettingsResponse(usersettings=[BringUserSettingsEntry(key='autoPush', value='ON'), BringUserSettingsEntry(key='purchaseStyle', value='grouped'), BringUserSettingsEntry(key='premiumHideSponsoredCategories', value='OFF'), BringUserSettingsEntry(key='premiumHideInspirationsBadge', value='OFF'), BringUserSettingsEntry(key='premiumHideOffersBadge', value='OFF'), BringUserSettingsEntry(key='premiumHideOffersOnMain', value='OFF'), BringUserSettingsEntry(key='defaultListUUID', value='xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxx'), BringUserSettingsEntry(key='discountActivatorOnMainEnabled', value='OFF'), BringUserSettingsEntry(key='onboardClient', value='android')], userlistsettings=[BringUserListSettingEntry(listUuid='00000000-0000-0000-0000-000000000000', usersettings=[BringUserSettingsEntry(key='listSectionOrder', value='["Früchte & Gemüse","Brot & Gebäck","Milch & Käse","Fleisch & Fisch","Zutaten & Gewürze","Fertig- & Tiefkühlprodukte","Getreideprodukte","Snacks & Süsswaren","Getränke & Tabak","Haushalt & Gesundheit","Pflege & Gesundheit","Tierbedarf","Baumarkt & Garten","Eigene Artikel"]'), BringUserSettingsEntry(key='listArticleLanguage', value='de-DE')])])
 # ---
 # name: TestGetList.test_get_list
   BringItemsResponse(uuid=UUID('00000000-0000-0000-0000-000000000000'), status='SHARED', items=Items(purchase=[BringPurchase(uuid=UUID('43bdd5a2-740a-4230-8b27-d0bbde886da7'), itemId='Paprika', specification='grün', attributes=[]), BringPurchase(uuid=UUID('2de9d1c0-c211-4129-b6c5-c1260c3fc735'), itemId='Zucchetti', specification='gelb', attributes=[])], recently=[BringPurchase(uuid=UUID('5681ed79-c8e4-4c8b-95ec-112999d016c0'), itemId='Paprika', specification='rot', attributes=[]), BringPurchase(uuid=UUID('01eea2cd-f433-4263-ad08-3d71317c4298'), itemId='Pouletbrüstli', specification='', attributes=[])]))

tests/conftest.py

@@ -8,7 +8,7 @@
 from bring_api.bring import Bring
 
 load_dotenv()
-UUID = "00000000-00000000-00000000-00000000"
+UUID = "00000000-0000-0000-0000-000000000000"
 
 BRING_LOGIN_RESPONSE = {
     "uuid": UUID,
@@ -24,8 +24,8 @@
 }
 
 BRING_USER_ACCOUNT_RESPONSE = {
-    "userUuid": "00000000-00000000-00000000-00000000",
-    "publicUserUuid": "00000000-00000000-00000000-00000000",
+    "userUuid": "00000000-0000-0000-0000-000000000000",
+    "publicUserUuid": "00000000-0000-0000-0000-000000000000",
     "email": "{email}",
     "emailVerified": True,
     "name": "{user_name}",
@@ -136,6 +136,69 @@
     "expires_in": 604799,
 }
 
+BRING_GET_ACTIVITY_RESPONSE = {
+    "timeline": [
+        {
+            "type": "LIST_ITEMS_CHANGED",
+            "content": {
+                "uuid": "673594a9-f92d-4cb6-adf1-d2f7a83207a4",
+                "purchase": [
+                    {
+                        "uuid": "658a3770-1a03-4ee0-94a6-10362a642377",
+                        "itemId": "Gurke",
+                        "specification": "",
+                        "attributes": [],
+                    }
+                ],
+                "recently": [
+                    {
+                        "uuid": "1ed22d3d-f19b-4530-a518-19872da3fd3e",
+                        "itemId": "Milch",
+                        "specification": "",
+                        "attributes": [],
+                    }
+                ],
+                "sessionDate": "2025-01-01T03:09:33.036Z",
+                "publicUserUuid": "98615d7e-0a7d-4a7e-8f73-a9cbb9f1bc32",
+            },
+        },
+        {
+            "type": "LIST_ITEMS_ADDED",
+            "content": {
+                "uuid": "9a16635c-dea2-4e00-904a-c5034f9cfecf",
+                "items": [
+                    {
+                        "uuid": "66a633a2-ae09-47bf-8845-3c0198480544",
+                        "itemId": "Joghurt",
+                        "specification": "",
+                        "attributes": [],
+                    },
+                ],
+                "sessionDate": "2025-01-01T02:54:57.656Z",
+                "publicUserUuid": "6743a171-247d-46d0-bc06-baf31194f949",
+            },
+        },
+        {
+            "type": "LIST_ITEMS_REMOVED",
+            "content": {
+                "uuid": "303dedf6-d4b2-4d25-a8cd-1c7967b84fcb",
+                "items": [
+                    {
+                        "uuid": "2ba8ddb6-01c6-4b0b-a89d-f3da6b291528",
+                        "itemId": "Tofu",
+                        "specification": "",
+                        "attributes": [],
+                    }
+                ],
+                "sessionDate": "2025-01-01T03:09:12.380Z",
+                "publicUserUuid": "6d79d10b-70b2-443f-9f7e-0b02e670c402",
+            },
+        },
+    ],
+    "timestamp": "2025-01-01T03:09:33.036Z",
+    "totalEvents": 2,
+}
+
 
 @pytest.fixture(name="headers")
 async def headers() -> str:

tests/test_bring.py

@@ -32,6 +32,7 @@
 )
 
 from .conftest import (
+    BRING_GET_ACTIVITY_RESPONSE,
     BRING_GET_ALL_ITEM_DETAILS_RESPONSE,
     BRING_GET_LIST_RESPONSE,
     BRING_LOAD_LISTS_RESPONSE,
@@ -1538,3 +1539,77 @@ async def test_value_error(self, bring):
 
         with pytest.raises(ValueError):
             await bring.set_list_article_language(UUID, "es-CO")
+
+
+class TestGetActivity:
+    """Tests for get_activity method."""
+
+    async def test_get_activity(
+        self,
+        bring,
+        mocked,
+        monkeypatch,
+        snapshot: SnapshotAssertion,
+    ):
+        """Test get_activity."""
+
+        mocked.get(
+            f"https://api.getbring.com/rest/v2/bringlists/{UUID}/activity",
+            status=HTTPStatus.OK,
+            payload=BRING_GET_ACTIVITY_RESPONSE,
+        )
+        monkeypatch.setattr(bring, "uuid", UUID)
+
+        activity = await bring.get_activity(uuid.UUID(UUID))
+
+        assert activity == snapshot
+
+    @pytest.mark.parametrize(
+        "exception",
+        [
+            asyncio.TimeoutError,
+            aiohttp.ClientError,
+        ],
+    )
+    async def test_request_exception(self, mocked, bring, exception):
+        """Test request exceptions."""
+
+        mocked.get(
+            f"https://api.getbring.com/rest/v2/bringlists/{UUID}/activity",
+            exception=exception,
+        )
+
+        with pytest.raises(BringRequestException):
+            await bring.get_activity(uuid.UUID(UUID))
+
+    async def test_auth_exception(self, mocked, bring):
+        """Test request exceptions."""
+
+        mocked.get(
+            f"https://api.getbring.com/rest/v2/bringlists/{UUID}/activity",
+            status=HTTPStatus.UNAUTHORIZED,
+            payload={"message": ""},
+        )
+
+        with pytest.raises(BringAuthException):
+            await bring.get_activity(uuid.UUID(UUID))
+
+    @pytest.mark.parametrize(
+        ("status", "exception"),
+        [
+            (HTTPStatus.OK, BringParseException),
+            (HTTPStatus.UNAUTHORIZED, BringAuthException),
+        ],
+    )
+    async def test_parse_exception(self, mocked, bring, status, exception):
+        """Test request exceptions."""
+
+        mocked.get(
+            f"https://api.getbring.com/rest/v2/bringlists/{UUID}/activity",
+            status=status,
+            body="not json",
+            content_type="application/json",
+        )
+
+        with pytest.raises(exception):
+            await bring.get_activity(uuid.UUID(UUID))