diff --git a/README.md b/README.md index d6544ec..32edc73 100644 --- a/README.md +++ b/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 diff --git a/armis_sdk/clients/assets_client.py b/armis_sdk/clients/assets_client.py index 9aed6b1..f3ec9f2 100644 --- a/armis_sdk/clients/assets_client.py +++ b/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, diff --git a/docs/index.md b/docs/index.md index 2fb5cc2..92c64ed 100644 --- a/docs/index.md +++ b/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/`). diff --git a/pyproject.toml b/pyproject.toml index a004f7a..f0071fa 100644 --- a/pyproject.toml +++ b/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" }, diff --git a/tests/armis_sdk/clients/assets_client_test.py b/tests/armis_sdk/clients/assets_client_test.py index 0e0e97e..0048cdc 100644 --- a/tests/armis_sdk/clients/assets_client_test.py +++ b/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",