refactor: simplify Users, factor out paginator (#49)

* Move client.users to property

* Refactor Users class

* Move pagination class to its own file

* Remove Resource classes (for now at least)

* Add coverage for final untested line (and fix a bug in it)

* Appease mypy

* rename to Paginator

src/posit/connect/client.py

@@ -7,8 +7,7 @@
 
 from .auth import Auth
 from .config import Config
-from .resources import CachedResources
-from .users import Users, User
+from .users import Users
 
 
 class Client:
@@ -33,8 +32,6 @@ def __init__(
         # Add error handling hooks to the session.
         session.hooks["response"].append(hooks.handle_errors)
 
-        # Initialize the Users instance.
-        self.users: CachedResources[User] = Users(config=self.config, session=session)
         # Store the Session object.
         self.session = session
 
@@ -47,6 +44,10 @@ def connect_version(self):
             self.server_settings = self.get("server_settings").json()
         return self.server_settings["version"]
 
+    @property
+    def users(self) -> Users:
+        return Users(config=self.config, session=self.session)
+
     def __del__(self):
         """
         Close the session when the Client instance is deleted.

src/posit/connect/client_test.py

@@ -25,31 +25,21 @@ def MockSession():
         yield mock
 
 
-@pytest.fixture
-def MockUsers():
-    with patch("posit.connect.client.Users") as mock:
-        yield mock
-
-
 class TestClient:
     def test_init(
         self,
         MockAuth: MagicMock,
         MockConfig: MagicMock,
         MockSession: MagicMock,
-        MockUsers: MagicMock,
     ):
         api_key = "foobar"
         url = "http://foo.bar/__api__"
         Client(api_key=api_key, url=url)
         MockAuth.assert_called_once_with(config=MockConfig.return_value)
         MockConfig.assert_called_once_with(api_key=api_key, url=url)
         MockSession.assert_called_once()
-        MockUsers.assert_called_once_with(
-            config=MockConfig.return_value, session=MockSession.return_value
-        )
 
-    def test__del__(self, MockAuth, MockConfig, MockSession, MockUsers):
+    def test__del__(self, MockAuth, MockConfig, MockSession):
         api_key = "foobar"
         url = "http://foo.bar/__api__"
         client = Client(api_key=api_key, url=url)

src/posit/connect/config_test.py

@@ -17,6 +17,7 @@ def test_get_api_key_empty():
         _get_api_key()
 
 
+@patch.dict("os.environ", clear=True)
 def test_get_api_key_miss():
     with pytest.raises(ValueError):
         _get_api_key()
@@ -34,6 +35,7 @@ def test_get_url_empty():
         _get_url()
 
 
+@patch.dict("os.environ", clear=True)
 def test_get_url_miss():
     with pytest.raises(ValueError):
         _get_url()

src/posit/connect/paginator.py

@@ -0,0 +1,49 @@
+from typing import List
+
+
+# The maximum page size supported by the API.
+_MAX_PAGE_SIZE = 500
+
+
+class Paginator:
+    """
+    Utility for consuming Connect APIs that have pagination
+
+    Usage:
+
+        pager = Paginator(client.session, url)
+        pager.get_all() # To return a list with results from all pages concatenated
+        pager.get_next_page() # To step through one page at a time
+    """
+
+    def __init__(
+        self, session, url, start_page: int = 1, page_size=_MAX_PAGE_SIZE
+    ) -> None:
+        self.session = session
+        self.url = url
+        self.page_number = start_page
+        self.page_size = page_size
+        # The API response will tell us how many total entries there are,
+        # but we don't know yet.
+        self.total = None
+        # We also want to track how many results we've seen so far
+        self.seen = 0
+
+    def get_all(self) -> List[dict]:
+        result = []
+        while self.total is None or self.seen < self.total:
+            result += self.get_next_page()
+            self.page_number += 1
+        return result
+
+    def get_next_page(self) -> List[dict]:
+        # Define query parameters for pagination.
+        params = {"page_number": self.page_number, "page_size": self.page_size}
+        # Send a GET request to the endpoint with the specified parameters.
+        response = self.session.get(self.url, params=params).json()
+        # On our first request, we won't have set the total yet, so do it
+        if self.total is None:
+            self.total = response["total"]
+        results = response["results"]
+        self.seen += len(results)
+        return results

src/posit/connect/users.py

@@ -1,20 +1,17 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import Iterator, Callable, List
+from typing import Callable, List, TypedDict
 
 from requests import Session
 
 from . import urls
 
 from .config import Config
-from .resources import Resources, Resource, CachedResources
+from .paginator import _MAX_PAGE_SIZE, Paginator
 
-# The maximum page size supported by the API.
-_MAX_PAGE_SIZE = 500
 
-
-class User(Resource, total=False):
+class User(TypedDict, total=False):
     guid: str
     email: str
     username: str
@@ -28,57 +25,30 @@ class User(Resource, total=False):
     locked: bool
 
 
-class CachedUsers(CachedResources[User]):
-    def find(self, filter: Callable[[User], bool] = lambda _: True) -> CachedUsers:
-        return CachedUsers([user for user in self if filter(user)])
-
-    def find_one(self, filter: Callable[[User], bool] = lambda _: True) -> User | None:
-        return next((user for user in self if filter(user)), None)
-
-    def get(self, id: str) -> User:
-        user = next((user for user in self if user["guid"] == id), None)
-        if user is None:
-            raise RuntimeError(f"failed to get user with id '{id}'")
-        return user
-
-
-class Users(CachedUsers, Resources[User]):
-    def __init__(
-        self, config: Config, session: Session, *, page_size=_MAX_PAGE_SIZE
-    ) -> None:
-        if page_size > _MAX_PAGE_SIZE:
-            raise ValueError(
-                f"page_size must be less than or equal to {_MAX_PAGE_SIZE}"
-            )
-
-        super().__init__(config.url)
+class Users:
+    def __init__(self, config: Config, session: Session) -> None:
         self.url = urls.append_path(config.url, "v1/users")
         self.config = config
         self.session = session
-        self.page_size = page_size
 
-    def fetch(self, index) -> tuple[Iterator[User] | None, bool]:
-        # Check if index is a multiple of page_size.
-        if (index % self.page_size) != 0:
-            #
-            raise ValueError(
-                f"index ({index}) must be a multiple of page size ({self.page_size})"
-            )
-        # Define the page number using 1-based indexing.
-        page_number = int(index / self.page_size) + 1
-        # Define query parameters for pagination.
-        params = {"page_number": page_number, "page_size": self.page_size}
-        # Send a GET request to the endpoint with the specified parameters.
-        response = self.session.get(self.url, params=params)
-        # Convert response to dict
-        json: dict = dict(response.json())
-        # Parse the JSON response and extract the results.
-        results: List[dict] = json["results"]
-        # Mark exhausted if the result size is less than the maximum page size.
-        exhausted = len(results) < self.page_size
-        # Create User objects from the results and return them as a list.
-        users: Iterator[User] = iter(User(**result) for result in results)
-        return (users, exhausted)
+    def find(
+        self, filter: Callable[[User], bool] = lambda _: True, page_size=_MAX_PAGE_SIZE
+    ) -> List[User]:
+        results = Paginator(self.session, self.url, page_size=page_size).get_all()
+        return [User(**user) for user in results if filter(User(**user))]
+
+    def find_one(
+        self, filter: Callable[[User], bool] = lambda _: True, page_size=_MAX_PAGE_SIZE
+    ) -> User | None:
+        pager = Paginator(self.session, self.url, page_size=page_size)
+        result = pager.get_next_page()
+        while pager.total is None or pager.seen < pager.total:
+            result = pager.get_next_page()
+            for u in result:
+                user = User(**u)
+                if filter(user):
+                    return user
+        return None
 
     def get(self, id: str) -> User:
         url = urls.append_path(self.url, id)