diff --git a/changelog.d/20260320_121518_sirosen_search_filter_roles.rst b/changelog.d/20260320_121518_sirosen_search_filter_roles.rst new file mode 100644 index 000000000..16a1d0774 --- /dev/null +++ b/changelog.d/20260320_121518_sirosen_search_filter_roles.rst @@ -0,0 +1,4 @@ +Added +----- + +- Added support for ``filter_roles`` as a parameter to ``SearchClient.index_list``. (:pr:`NUMBER`) diff --git a/src/globus_sdk/services/search/client.py b/src/globus_sdk/services/search/client.py index 88d1a49ea..1628f05c9 100644 --- a/src/globus_sdk/services/search/client.py +++ b/src/globus_sdk/services/search/client.py @@ -1,11 +1,12 @@ from __future__ import annotations import logging +import sys import typing as t import uuid from globus_sdk import client, paging, response -from globus_sdk._internal.remarshal import strseq_listify +from globus_sdk._internal.remarshal import commajoin, strseq_listify from globus_sdk._missing import MISSING, MissingType from globus_sdk.scopes import SearchScopes @@ -13,8 +14,15 @@ from .errors import SearchAPIError from .response import IndexListResponse +if sys.version_info >= (3, 10): + from typing import TypeAlias +else: + from typing_extensions import TypeAlias + log = logging.getLogger(__name__) +_VALID_ROLE_NAMES_T: TypeAlias = t.Literal["owner", "admin", "writer"] + class SearchClient(client.BaseClient): r""" @@ -229,12 +237,18 @@ def get_index( def index_list( self, + filter_roles: ( + _VALID_ROLE_NAMES_T | t.Iterable[_VALID_ROLE_NAMES_T] | MissingType + ) = MISSING, *, query_params: dict[str, t.Any] | None = None, ) -> response.IterableResponse: """ Get a list of indices on which the caller has permissions. + :param filter_roles: An iterable of roles to use to filter the listing. By + default, all indices where the user has a role are returned. + Valid values are ``owner``, ``admin``, and ``writer``. :param query_params: additional parameters to pass as query params .. tab-set:: @@ -260,6 +274,7 @@ def index_list( :ref: search/reference/index_list/ """ # noqa: E501 log.debug("SearchClient.index_list()") + query_params = {"filter_roles": commajoin(filter_roles), **(query_params or {})} return IndexListResponse(self.get("/v1/index_list", query_params=query_params)) # diff --git a/tests/functional/services/search/test_index_list.py b/tests/functional/services/search/test_index_list.py index be2d2b494..fb24286b2 100644 --- a/tests/functional/services/search/test_index_list.py +++ b/tests/functional/services/search/test_index_list.py @@ -1,4 +1,10 @@ -from globus_sdk.testing import load_response +import dataclasses +import urllib.parse + +import pytest + +from globus_sdk._missing import MISSING +from globus_sdk.testing import get_last_request, load_response def test_search_index_list(client): @@ -24,3 +30,54 @@ def test_search_index_list_is_iterable(client): index_list = list(res) assert len(index_list) == len(index_ids) assert [i["id"] for i in index_list] == index_ids + + +_OMIT = object() + + +@dataclasses.dataclass +class FilterRoleParam: + name: str + value: object + expect_parsed_param: object = None + + @property + def missing(self) -> bool: + return self.value in (_OMIT, MISSING) + + @property + def call_kwargs(self) -> dict[str, object]: + return {} if self.value is _OMIT else {"filter_roles": self.value} + + def __str__(self) -> str: + return self.name + + +@pytest.mark.parametrize( + "filter_param", + [ + FilterRoleParam("omitted", _OMIT), + FilterRoleParam("missing", MISSING), + FilterRoleParam("ownerstr", "owner", ["owner"]), + FilterRoleParam("adminstr", "admin", ["admin"]), + FilterRoleParam("writerstr", "writer", ["writer"]), + FilterRoleParam("tuple", ("owner", "admin"), ["owner,admin"]), + FilterRoleParam("list", ["admin", "writer"], ["admin,writer"]), + FilterRoleParam("duplicates", ("admin", "admin"), ["admin,admin"]), + # this isn't a real role, but it should be passed through as though it were + # -- this value can be a typing time error but never a runtime error + FilterRoleParam("unknown_str", "ambassador", ["ambassador"]), + ], + ids=str, +) +def test_search_index_list_encodes_filter_roles_as_expected(client, filter_param): + load_response(client.index_list).metadata + res = client.index_list(**filter_param.call_kwargs) + assert res.http_status == 200 + + req = get_last_request() + parsed_qs = urllib.parse.parse_qs(urllib.parse.urlparse(req.url).query) + if filter_param.missing: + assert "filter_roles" not in parsed_qs + else: + assert parsed_qs["filter_roles"] == filter_param.expect_parsed_param