Add asset filtering methods (list_by_boundary_id, list_by_site_id, list_by_multiple)

README.md

@@ -66,8 +66,8 @@ Once you have an instance of `ArmisSdk`, you can start interacting with the vari
 
 > [!NOTE]
 > Note that all functions in this SDK that eventually make HTTP requests are asynchronous.
-> 
-> However, for convenience, all public asynchronous functions can also be executed in a synchronous way. 
+>
+> However, for convenience, all public asynchronous functions can also be executed in a synchronous way.
 
 For example, if you want to update a site's location:
 ```python

armis_sdk/clients/assets_client.py

@@ -53,7 +53,7 @@ async def list_by_asset_id(
             Assets of the specified class matching the provided identifiers.
 
         Example:
-            ```python linenums="1" hl_lines="13 17"
+            ```python linenums="1" hl_lines="14 18"
             import asyncio
 
             from armis_sdk.clients.assets_client import AssetsClient
@@ -78,6 +78,8 @@ async def main():
             asyncio.run(main())
             ```
         """
+        if not asset_ids:
+            raise ArmisError("asset_ids must not be empty")
         filter_ = {
             "filter_criteria": "ASSET_ID",
             "asset_ids": asset_ids,
@@ -86,6 +88,44 @@ async def main():
         async for item in self._list_assets(asset_class, fields, filter_):
             yield item
 
+    async def list_by_boundary_id(
+        self,
+        asset_class: type[AssetT],
+        boundary_ids: list[int],
+        fields: list[str] | None = None,
+    ) -> AsyncIterator[AssetT]:
+        """List assets by boundary ID.
+
+        Args:
+            asset_class: The asset class to list. Must inherit from [Asset][armis_sdk.entities.asset.Asset].
+            boundary_ids: A list of boundary IDs to filter by.
+            fields: Optional list of fields to retrieve. If None, all non-custom fields are retrieved.
+
+        Yields:
+            Assets of the specified class belonging to any of the provided boundaries.
+
+        Example:
+            ```python linenums="1" hl_lines="10"
+            import asyncio
+
+            from armis_sdk.clients.assets_client import AssetsClient
+            from armis_sdk.entities.device import Device
+
+
+            async def main():
+                assets_client = AssetsClient()
+
+                async for device in assets_client.list_by_boundary_id(Device, [1, 2, 3]):
+                    print(device)
+
+
+            asyncio.run(main())
+            ```
+        """
+        filter_ = self._build_boundary_id_filter(boundary_ids)
+        async for item in self._list_assets(asset_class, fields, filter_):
+            yield item
+
     async def list_by_last_seen(
         self,
         asset_class: type[AssetT],
@@ -106,7 +146,7 @@ async def list_by_last_seen(
             ArmisError: If last_seen is neither datetime nor timedelta.
 
         Example:
-            ```python linenums="1" hl_lines="11 15"
+            ```python linenums="1" hl_lines="12 16"
             import asyncio
             import datetime
 
@@ -129,15 +169,115 @@ async def main():
             asyncio.run(main())
             ```
         """
-        filter_: dict[str, str | int] = {"filter_criteria": "LAST_SEEN"}
+        filter_ = self._build_last_seen_filter(last_seen)
+        async for item in self._list_assets(asset_class, fields, filter_):
+            yield item
 
-        if isinstance(last_seen, datetime.datetime):
-            filter_["last_seen_ge"] = last_seen.isoformat()
-        elif isinstance(last_seen, datetime.timedelta):
-            filter_["last_seen_seconds"] = int(last_seen.total_seconds())
-        else:
-            raise ArmisError(f"Invalid 'last_seen' type {type(last_seen)}")
+    async def list_by_multiple(
+        self,
+        asset_class: type[AssetT],
+        last_seen: datetime.datetime | datetime.timedelta | None = None,
+        site_ids: list[int] | None = None,
+        boundary_ids: list[int] | None = None,
+        fields: list[str] | None = None,
+    ) -> AsyncIterator[AssetT]:
+        """List assets matching multiple filter criteria simultaneously (AND logic).
+
+        At least one of `last_seen`, `site_ids`, or `boundary_ids` must be provided.
+        Each criterion that is provided is applied as an AND condition.
 
+        Args:
+            asset_class: The asset class to list. Must inherit from [Asset][armis_sdk.entities.asset.Asset].
+            last_seen: Either a datetime (assets seen on or after this time) or timedelta (assets seen within this duration).
+            site_ids: A list of site IDs to filter by.
+            boundary_ids: A list of boundary IDs to filter by.
+            fields: Optional list of fields to retrieve. If None, all non-custom fields are retrieved.
+
+        Yields:
+            Assets of the specified class matching all provided criteria.
+
+        Raises:
+            ArmisError: If no filter criteria are provided, or if last_seen is an invalid type.
+
+        Example:
+            ```python linenums="1" hl_lines="11-15"
+            import asyncio
+            import datetime
+
+            from armis_sdk.clients.assets_client import AssetsClient
+            from armis_sdk.entities.device import Device
+
+
+            async def main():
+                assets_client = AssetsClient()
+
+                async for device in assets_client.list_by_multiple(
+                    Device,
+                    site_ids=[1, 2],
+                    last_seen=datetime.timedelta(hours=1),
+                ):
+                    print(device)
+
+
+            asyncio.run(main())
+            ```
+        """
+        filters = []
+
+        if last_seen is not None:
+            filters.append(self._build_last_seen_filter(last_seen))
+
+        if site_ids is not None:
+            filters.append(self._build_site_id_filter(site_ids))
+
+        if boundary_ids is not None:
+            filters.append(self._build_boundary_id_filter(boundary_ids))
+
+        if not filters:
+            raise ArmisError("At least one of filter must be provided")
+
+        filter_ = {
+            "filter_criteria": "MULTIPLE",
+            "filters": filters,
+        }
+        async for item in self._list_assets(asset_class, fields, filter_):
+            yield item
+
+    async def list_by_site_id(
+        self,
+        asset_class: type[AssetT],
+        site_ids: list[int],
+        fields: list[str] | None = None,
+    ) -> AsyncIterator[AssetT]:
+        """List assets by site ID.
+
+        Args:
+            asset_class: The asset class to list. Must inherit from [Asset][armis_sdk.entities.asset.Asset].
+            site_ids: A list of site IDs to filter by.
+            fields: Optional list of fields to retrieve. If None, all non-custom fields are retrieved.
+
+        Yields:
+            Assets of the specified class belonging to any of the provided sites.
+
+        Example:
+            ```python linenums="1" hl_lines="10"
+            import asyncio
+
+            from armis_sdk.clients.assets_client import AssetsClient
+            from armis_sdk.entities.device import Device
+
+
+            async def main():
+                assets_client = AssetsClient()
+
+                async for device in assets_client.list_by_site_id(Device, [1, 2, 3]):
+                    print(device)
+
+
+            asyncio.run(main())
+            ```
+        """
+        filter_ = self._build_site_id_filter(site_ids)
         async for item in self._list_assets(asset_class, fields, filter_):
             yield item
 
@@ -151,7 +291,7 @@ async def list_fields(self, asset_class: type[AssetT]) -> AsyncIterator[AssetFie
             Field descriptions including field name, type, and other metadata.
 
         Example:
-            ```python linenums="1" hl_lines="9"
+            ```python linenums="1" hl_lines="10"
             import asyncio
 
             from armis_sdk.clients.assets_client import AssetsClient
@@ -249,6 +389,29 @@ async def main():
             if errors:
                 raise BulkUpdateError(errors)
 
+    @staticmethod
+    def _build_boundary_id_filter(boundary_ids: list[int]) -> dict:
+        if not boundary_ids:
+            raise ArmisError("boundary_ids must not be empty")
+        return {"filter_criteria": "BOUNDARY_ID", "boundary_ids": boundary_ids}
+
+    @staticmethod
+    def _build_last_seen_filter(last_seen: datetime.datetime | datetime.timedelta) -> dict:
+        filter_: dict[str, str | int] = {"filter_criteria": "LAST_SEEN"}
+        if isinstance(last_seen, datetime.datetime):
+            filter_["last_seen_ge"] = last_seen.isoformat()
+        elif isinstance(last_seen, datetime.timedelta):
+            filter_["last_seen_seconds"] = int(last_seen.total_seconds())
+        else:
+            raise ArmisError(f"Invalid 'last_seen' type {type(last_seen)}")
+        return filter_
+
+    @staticmethod
+    def _build_site_id_filter(site_ids: list[int]) -> dict:
+        if not site_ids:
+            raise ArmisError("site_ids must not be empty")
+        return {"filter_criteria": "SITE_ID", "site_ids": site_ids}
+
     @classmethod
     def _create_bulk_update_request(
         cls,

docs/index.md

@@ -9,7 +9,7 @@ pip install armis_sdk
 ```
 ## Usage
 
-All interaction with the SDK happens through the [ArmisSdk][armis_sdk.core.armis_sdk.ArmisSdk] class. 
+All interaction with the SDK happens through the [ArmisSdk][armis_sdk.core.armis_sdk.ArmisSdk] class.
 You'll need five things:
 
 1. **Audience**: The url of the tenant you want to interact with, including trailing slash (e.g. `https://acme.armis.com/`).

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "armis_sdk"
-version = "1.1.3"
+version = "1.2.0"
 description = "The Armis SDK is a package that encapsulates common use-cases for interacting with the Armis platform."
 authors = [
     { name = "Shai Lachmanovich", email = "shai@armis.com" },

tests/armis_sdk/clients/assets_client_test.py

@@ -200,6 +200,133 @@ async def test_list_by_asset_id_invalid_fields():
             pass
 
 
+async def test_list_by_boundary_id(httpx_mock: pytest_httpx.HTTPXMock):
+    httpx_mock.add_response(
+        url="https://api.armis.com/v3/assets/_search",
+        method="POST",
+        match_json={
+            "limit": 100,
+            "asset_type": "DEVICE",
+            "fields": assets_test_data.ALL_DEVICE_FIELDS,
+            "filter": {"filter_criteria": "BOUNDARY_ID", "boundary_ids": [1, 2, 3]},
+        },
+        json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
+    )
+
+    assets_client = AssetsClient()
+    devices = [device async for device in assets_client.list_by_boundary_id(Device, [1, 2, 3])]
+
+    assert devices == [assets_test_data.MOCK_DEVICE_FULL]
+
+
+@pytest.mark.parametrize(
+    ["kwargs", "expected_filters"],
+    [
+        (
+            {"boundary_ids": [1, 2, 3]},
+            [{"filter_criteria": "BOUNDARY_ID", "boundary_ids": [1, 2, 3]}],
+        ),
+        (
+            {"site_ids": [1, 2, 3]},
+            [{"filter_criteria": "SITE_ID", "site_ids": [1, 2, 3]}],
+        ),
+        (
+            {"last_seen": datetime.datetime(2025, 12, 3)},
+            [{"filter_criteria": "LAST_SEEN", "last_seen_ge": "2025-12-03T00:00:00"}],
+        ),
+        (
+            {"last_seen": datetime.timedelta(hours=1)},
+            [{"filter_criteria": "LAST_SEEN", "last_seen_seconds": 3600}],
+        ),
+        (
+            {"last_seen": datetime.timedelta(hours=1), "site_ids": [1, 2], "boundary_ids": [3, 4]},
+            [
+                {"filter_criteria": "LAST_SEEN", "last_seen_seconds": 3600},
+                {"filter_criteria": "SITE_ID", "site_ids": [1, 2]},
+                {"filter_criteria": "BOUNDARY_ID", "boundary_ids": [3, 4]},
+            ],
+        ),
+    ],
+)
+async def test_list_by_multiple(httpx_mock: pytest_httpx.HTTPXMock, kwargs, expected_filters):
+    httpx_mock.add_response(
+        url="https://api.armis.com/v3/assets/_search",
+        method="POST",
+        match_json={
+            "limit": 100,
+            "asset_type": "DEVICE",
+            "fields": assets_test_data.ALL_DEVICE_FIELDS,
+            "filter": {"filter_criteria": "MULTIPLE", "filters": expected_filters},
+        },
+        json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
+    )
+
+    assets_client = AssetsClient()
+    devices = [device async for device in assets_client.list_by_multiple(Device, **kwargs)]
+
+    assert devices == [assets_test_data.MOCK_DEVICE_FULL]
+
+
+async def test_list_by_multiple_no_filters():
+    assets_client = AssetsClient()
+
+    with pytest.raises(ArmisError, match="At least one of"):
+        async for _ in assets_client.list_by_multiple(Device):
+            pass
+
+
+@pytest.mark.parametrize(
+    ["kwargs", "expected_error"],
+    [
+        (
+            {"last_seen": "2025-12-03"},
+            r"Invalid 'last_seen' type",
+        ),
+        (
+            {"last_seen": 3600},
+            r"Invalid 'last_seen' type",
+        ),
+        (
+            {"boundary_ids": []},
+            "boundary_ids must not be empty",
+        ),
+        (
+            {"site_ids": []},
+            "site_ids must not be empty",
+        ),
+        (
+            {"site_ids": [1, 2], "fields": ["device_id", "foo", "bar"]},
+            "The following fields are not supported with this operation: 'foo', 'bar'",
+        ),
+    ],
+)
+async def test_list_by_multiple_invalid_input(kwargs, expected_error):
+    assets_client = AssetsClient()
+
+    with pytest.raises(ArmisError, match=expected_error):
+        async for _ in assets_client.list_by_multiple(Device, **kwargs):
+            pass
+
+
+async def test_list_by_site_id(httpx_mock: pytest_httpx.HTTPXMock):
+    httpx_mock.add_response(
+        url="https://api.armis.com/v3/assets/_search",
+        method="POST",
+        match_json={
+            "limit": 100,
+            "asset_type": "DEVICE",
+            "fields": assets_test_data.ALL_DEVICE_FIELDS,
+            "filter": {"filter_criteria": "SITE_ID", "site_ids": [1, 2, 3]},
+        },
+        json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
+    )
+
+    assets_client = AssetsClient()
+    devices = [device async for device in assets_client.list_by_site_id(Device, [1, 2, 3])]
+
+    assert devices == [assets_test_data.MOCK_DEVICE_FULL]
+
+
 async def test_update(httpx_mock: pytest_httpx.HTTPXMock):
     httpx_mock.add_response(
         url="https://api.armis.com/v3/assets/_bulk",