Merge pull request #901 from CitrineInformatics/as_admin-listing

as_admin listing for users and teams

Author: robbtraister

src/citrine/__version__.py

@@ -1 +1 @@
-__version__ = "2.40.0"
+__version__ = "2.41.0"

src/citrine/_rest/admin_collection.py

@@ -0,0 +1,44 @@
+from functools import partial
+from typing import Iterator
+
+from citrine._rest.collection import Collection, ResourceType
+
+
+class AdminCollection(Collection[ResourceType]):
+    """Abstract class for representing collections of REST resources with as_admin access."""
+
+    def list(
+        self, *, per_page: int = 100, as_admin: bool = False
+    ) -> Iterator[ResourceType]:
+        """
+        Paginate over the elements of the collection.
+
+        Leaving page and per_page as default values will yield all elements in the
+        collection, paginating over all available pages.
+
+        Parameters
+        ---------
+        per_page: int, optional
+            Max number of results to return per page. Default is 100.  This parameter
+            is used when making requests to the backend service.  If the page parameter
+            is specified it limits the maximum number of elements in the response.
+
+        as_admin: bool, optional
+            Whether this request should be made as an admin (returning all teams,
+            rather than only those to which the user belongs).
+
+        Returns
+        -------
+        Iterator[ResourceType]
+            An iterator that can be used to loop over all the resources in this collection.
+            Use list() to force evaluation of all results into an in-memory list.
+
+        """
+        fetcher = partial(
+            self._fetch_page, additional_params={"as_admin": "true"} if as_admin else {}
+        )
+        return self._paginator.paginate(
+            page_fetcher=fetcher,
+            collection_builder=self._build_collection_elements,
+            per_page=per_page,
+        )

src/citrine/resources/team.py

@@ -2,7 +2,7 @@
 from typing import Optional, Union, List
 from uuid import UUID
 
-from citrine._rest.collection import Collection
+from citrine._rest.admin_collection import AdminCollection
 from citrine._rest.resource import Resource, ResourceTypeEnum
 from citrine._serialization import properties
 from citrine._session import Session
@@ -394,7 +394,7 @@ def table_definition_ids(self) -> TeamResourceIDs:
                                resource_type=ResourceTypeEnum.TABLE_DEFINITION.value)
 
 
-class TeamCollection(Collection[Team]):
+class TeamCollection(AdminCollection[Team]):
     """
     Represents the collection of all teams as well as the resources belonging to it.
 

src/citrine/resources/user.py

@@ -1,7 +1,7 @@
 """Resources that represent both individual and collections of users."""
 from typing import Optional
 
-from citrine._rest.collection import Collection
+from citrine._rest.admin_collection import AdminCollection
 from citrine._rest.resource import Resource, ResourceTypeEnum
 from citrine._serialization import properties
 from citrine._session import Session
@@ -59,7 +59,7 @@ def get(self):
         raise NotImplementedError("Get Not Implemented in Citrine Platform")
 
 
-class UserCollection(Collection[User]):
+class UserCollection(AdminCollection[User]):
     """Represents the collection of all users."""
 
     _path_template = '/users'

tests/resources/test_team.py

@@ -148,6 +148,25 @@ def test_list_teams(collection, session):
     assert 5 == len(teams)
 
 
+def test_list_teams_as_admin(collection, session):
+    # Given
+    teams_data = TeamDataFactory.create_batch(5)
+    session.set_response({"teams": teams_data})
+
+    # When
+    teams = list(collection.list(as_admin=True))
+
+    # Then
+    assert 1 == session.num_calls
+    expected_call = FakeCall(
+        method="GET",
+        path="/teams",
+        params={"per_page": 100, "page": 1, "as_admin": "true"},
+    )
+    assert expected_call == session.last_call
+    assert 5 == len(teams)
+
+
 def test_update_team(collection: TeamCollection, team, session):
     team.name = "updated name"
     session.set_response({'team': team.dump()})

tests/resources/test_user.py

@@ -98,6 +98,26 @@ def test_list_users(collection, session):
     assert len(users) == 5
 
 
+def test_list_users_as_admin(collection, session):
+    # Given
+    user_data = UserDataFactory.create_batch(5)
+    session.set_response({'users': user_data})
+
+    # When
+    users = list(collection.list(as_admin=True))
+
+    # Then
+    assert 1 == session.num_calls
+    expected_call = FakeCall(
+        method='GET',
+        path='/users',
+        params={'per_page': 100, 'page': 1, 'as_admin': 'true'}
+    )
+
+    assert expected_call == session.last_call
+    assert len(users) == 5
+
+
 def test_get_users(collection, session):
     # Given
     uid = '151199ec-e9aa-49a1-ac8e-da722aaf74c4'