Skip to content

Pass limit parameter in the correct location during assets pagination #58

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
shaiarmis merged 1 commit into from
Apr 27, 2026
Merged

Pass limit parameter in the correct location during assets pagination #58

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion armis_sdk/clients/assets_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -501,7 +501,7 @@ async def _list_assets(
"fields": fields,
"filter": filter_,
}
async for item in self._armis_client.list("/v3/assets/_search", body=body, after_location="filter"):
async for item in self._armis_client.list("/v3/assets/_search", body=body, pagination_location="filter"):
yield asset_class.from_search_result(item)

@classmethod
Expand Down
19 changes: 14 additions & 5 deletions armis_sdk/core/armis_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,13 +89,18 @@ def client(self, retries: int | None = None, backoff: float | None = None):
trust_env=True,
)

async def list(self, url: str, body: dict | None = None, after_location: str | None = None) -> AsyncIterator[dict]:
async def list(
self,
url: str,
body: dict | None = None,
pagination_location: str | None = None,
) -> AsyncIterator[dict]:
"""List all items from a paginated endpoint.

Args:
url (str): The relative endpoint URL.
body (dict): Payload to send as POST request.
after_location (str): The nested object location to use for pagination.
pagination_location (str): The nested object location to use for pagination.

Returns:
An (async) iterator of `dict`s.
Expand Down Expand Up @@ -123,7 +128,11 @@ async def main():
"""
page_size = int(os.getenv(ARMIS_PAGE_SIZE, str(DEFAULT_PAGE_LENGTH)))
async with self.client() as client:
params = {"limit": page_size, **(body or {})}
params = {**(body or {})}
if pagination_location:
params[pagination_location]["limit"] = page_size
else:
params["limit"] = page_size
while True:
if body:
response = await client.post(url, json=params)
Expand All @@ -134,8 +143,8 @@ async def main():
for item in items:
yield item
if next_ := data.get("next"):
if after_location:
params[after_location]["after"] = next_
if pagination_location:
params[pagination_location]["after"] = next_
else:
params["after"] = next_
else:
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[project]
name = "armis_sdk"
version = "1.2.1"
version = "1.2.2"
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" },
Expand Down
53 changes: 35 additions & 18 deletions tests/armis_sdk/clients/assets_client_test.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,12 +25,12 @@ async def test_list_by_last_seen_datetime(httpx_mock: pytest_httpx.HTTPXMock):
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": "LAST_SEEN",
"last_seen_ge": "2025-12-03T00:00:00",
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
Expand All @@ -50,12 +50,12 @@ async def test_list_by_last_seen_datetime_explicit_fields(
url="https://api.armis.com/v3/assets/_search",
method="POST",
match_json={
"limit": 100,
"asset_type": "DEVICE",
"fields": ["brand", "custom.MyField1", "custom.MyField2", "purdue_level"],
"filter": {
"filter_criteria": "LAST_SEEN",
"last_seen_ge": "2025-12-03T00:00:00",
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_PARTIAL_RAW_DATA}]},
Expand All @@ -74,10 +74,13 @@ async def test_list_by_last_seen_timedelta(httpx_mock: pytest_httpx.HTTPXMock):
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": "LAST_SEEN", "last_seen_seconds": 3600},
"filter": {
"filter_criteria": "LAST_SEEN",
"last_seen_seconds": 3600,
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
)
Expand All @@ -96,10 +99,13 @@ async def test_list_by_last_seen_timedelta_explicit_fields(
url="https://api.armis.com/v3/assets/_search",
method="POST",
match_json={
"limit": 100,
"asset_type": "DEVICE",
"fields": ["brand", "custom.MyField1", "custom.MyField2", "purdue_level"],
"filter": {"filter_criteria": "LAST_SEEN", "last_seen_seconds": 3600},
"filter": {
"filter_criteria": "LAST_SEEN",
"last_seen_seconds": 3600,
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_PARTIAL_RAW_DATA}]},
)
Expand Down Expand Up @@ -130,13 +136,13 @@ async def test_list_by_asset_id(httpx_mock: pytest_httpx.HTTPXMock):
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": "ASSET_ID",
"asset_id_source": "IPV4_ADDRESS",
"asset_ids": ["1.1.1.1"],
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
Expand All @@ -161,13 +167,13 @@ async def test_list_by_asset_id_explicit_fields(httpx_mock: pytest_httpx.HTTPXMo
url="https://api.armis.com/v3/assets/_search",
method="POST",
match_json={
"limit": 100,
"asset_type": "DEVICE",
"fields": ["brand", "custom.MyField1", "custom.MyField2", "purdue_level"],
"filter": {
"filter_criteria": "ASSET_ID",
"asset_id_source": "IPV4_ADDRESS",
"asset_ids": ["1.1.1.1"],
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_PARTIAL_RAW_DATA}]},
Expand Down Expand Up @@ -205,10 +211,13 @@ async def test_list_by_boundary_id(httpx_mock: pytest_httpx.HTTPXMock):
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]},
"filter": {
"filter_criteria": "BOUNDARY_ID",
"boundary_ids": [1, 2, 3],
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
)
Expand Down Expand Up @@ -253,10 +262,13 @@ async def test_list_by_multiple(httpx_mock: pytest_httpx.HTTPXMock, kwargs, expe
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},
"filter": {
"filter_criteria": "MULTIPLE",
"filters": expected_filters,
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
)
Expand Down Expand Up @@ -313,10 +325,13 @@ async def test_list_by_site_id(httpx_mock: pytest_httpx.HTTPXMock):
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]},
"filter": {
"filter_criteria": "SITE_ID",
"site_ids": [1, 2, 3],
"limit": 100,
},
},
json={"items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
)
Expand Down Expand Up @@ -499,21 +514,23 @@ async def test_list_assets_pagination(monkeypatch, httpx_mock: pytest_httpx.HTTP
url="https://api.armis.com/v3/assets/_search",
method="POST",
match_json={
"limit": 1,
"asset_type": "DEVICE",
"fields": assets_test_data.ALL_DEVICE_FIELDS,
"filter": {"filter_criteria": "LAST_SEEN", "last_seen_seconds": 3600},
"filter": {
"filter_criteria": "LAST_SEEN",
"last_seen_seconds": 3600,
"limit": 1,
},
},
json={"next": 2, "items": [{"asset_id": 1, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
)
httpx_mock.add_response(
url="https://api.armis.com/v3/assets/_search",
method="POST",
match_json={
"limit": 1,
"asset_type": "DEVICE",
"fields": assets_test_data.ALL_DEVICE_FIELDS,
"filter": {"filter_criteria": "LAST_SEEN", "last_seen_seconds": 3600, "after": 2},
"filter": {"filter_criteria": "LAST_SEEN", "last_seen_seconds": 3600, "limit": 1, "after": 2},
},
json={"next": None, "items": [{"asset_id": 2, "fields": assets_test_data.MOCK_DEVICE_FULL_RAW_DATA}]},
)
Expand Down
Loading