Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 4 additions & 0 deletions changelog.d/20260320_121518_sirosen_search_filter_roles.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
Added
-----

- Added support for ``filter_roles`` as a parameter to ``SearchClient.index_list``. (:pr:`NUMBER`)
17 changes: 16 additions & 1 deletion src/globus_sdk/services/search/client.py
Original file line number Diff line number Diff line change
@@ -1,20 +1,28 @@
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

from .data import SearchQueryV1, SearchScrollQuery
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"""
Expand Down Expand Up @@ -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::
Expand All @@ -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))

#
Expand Down
59 changes: 58 additions & 1 deletion tests/functional/services/search/test_index_list.py
Original file line number Diff line number Diff line change
@@ -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):
Expand All @@ -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
Loading