Merge pull request #101 from NimishaShrivastava-dev/refactor-sshkey-list-iterator

Ssh-iterator pattern added

Author: isivaselvan

.gitignore

@@ -137,3 +137,4 @@ protos.bin
 
 # Local RSA keys
 .local/*
+

examples/ssh_keys.py

@@ -99,10 +99,11 @@ def main():
     try:
         # 1. List existing SSH keys
         print("\n1. Listing SSH keys...")
-        ssh_keys = client.ssh_keys.list(TFE_ORG)
-        print(f"Found {len(ssh_keys.items)} SSH keys:")
-        for key in ssh_keys.items:
+        ssh_keys_count = 0
+        for key in client.ssh_keys.list(TFE_ORG):
             print(f"- ID: {key.id}, Name: {key.name}")
+            ssh_keys_count += 1
+        print(f"Found {ssh_keys_count} SSH keys")
 
         # 2. Create a new SSH key
         print("\n2. Creating a new SSH key...")
@@ -132,16 +133,17 @@ def main():
 
         # 6. Verify deletion by listing again
         print("\n6. Verifying deletion...")
-        ssh_keys_after = client.ssh_keys.list(TFE_ORG)
-        print(f"SSH keys after deletion: {len(ssh_keys_after.items)}")
+        ssh_keys_after_count = sum(1 for _ in client.ssh_keys.list(TFE_ORG))
+        print(f"SSH keys after deletion: {ssh_keys_after_count}")
 
-        # 7. Demonstrate pagination with options
-        print("\n7. Demonstrating pagination options...")
+        # 7. Demonstrate iterator with pagination options
+        print("\n7. Demonstrating iterator with pagination options...")
         list_options = SSHKeyListOptions(page_size=5, page_number=1)
-        paginated_keys = client.ssh_keys.list(TFE_ORG, list_options)
-        print(f"Page 1 with page size 5: {len(paginated_keys.items)} keys")
-        print(f"Total pages: {paginated_keys.total_pages}")
-        print(f"Total count: {paginated_keys.total_count}")
+        paginated_count = 0
+        for key in client.ssh_keys.list(TFE_ORG, list_options):
+            paginated_count += 1
+            print(f"  - {key.name}")
+        print(f"Listed {paginated_count} keys with pagination options")
 
         print("\n SSH Keys API example completed successfully!")
 

src/pytfe/client.py

@@ -69,6 +69,7 @@ def __init__(self, config: TFEConfig | None = None):
         self.plans = Plans(self._transport)
         self.organizations = Organizations(self._transport)
         self.organization_memberships = OrganizationMemberships(self._transport)
+
         self.projects = Projects(self._transport)
         self.variables = Variables(self._transport)
         self.variable_sets = VariableSets(self._transport)

src/pytfe/models/__init__.py

@@ -287,7 +287,6 @@
 from .ssh_key import (
     SSHKey,
     SSHKeyCreateOptions,
-    SSHKeyList,
     SSHKeyListOptions,
     SSHKeyUpdateOptions,
 )
@@ -395,7 +394,6 @@
     # SSH keys
     "SSHKey",
     "SSHKeyCreateOptions",
-    "SSHKeyList",
     "SSHKeyListOptions",
     "SSHKeyUpdateOptions",
     # Reserved tag keys

src/pytfe/resources/ssh_keys.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Iterator
 from typing import Any
 
 from ..errors import (
@@ -9,7 +10,6 @@
 from ..models.ssh_key import (
     SSHKey,
     SSHKeyCreateOptions,
-    SSHKeyList,
     SSHKeyListOptions,
     SSHKeyUpdateOptions,
 )
@@ -22,37 +22,17 @@ class SSHKeys(_Service):
 
     def list(
         self, organization: str, options: SSHKeyListOptions | None = None
-    ) -> SSHKeyList:
+    ) -> Iterator[SSHKey]:
         """List SSH keys for the given organization."""
         if not valid_string_id(organization):
             raise InvalidOrgError()
 
-        params = (
-            options.model_dump(by_alias=True, exclude_none=True) if options else None
-        )
-
-        r = self.t.request(
-            "GET",
-            f"/api/v2/organizations/{organization}/ssh-keys",
-            params=params,
-        )
-
-        jd = r.json()
-        items = []
-        meta = jd.get("meta", {})
-        pagination = meta.get("pagination", {})
-
-        for d in jd.get("data", []):
-            items.append(self._parse_ssh_key(d))
-
-        return SSHKeyList(
-            items=items,
-            current_page=pagination.get("current-page"),
-            total_pages=pagination.get("total-pages"),
-            prev_page=pagination.get("prev-page"),
-            next_page=pagination.get("next-page"),
-            total_count=pagination.get("total-count"),
-        )
+        params = options.model_dump(by_alias=True, exclude_none=True) if options else {}
+        path = f"/api/v2/organizations/{organization}/ssh-keys"
+        for item in self._list(path, params=params):
+            attrs = item.get("attributes", {})
+            attrs["id"] = item.get("id")
+            yield SSHKey.model_validate(attrs)
 
     def create(self, organization: str, options: SSHKeyCreateOptions) -> SSHKey:
         """Create a new SSH key for the given organization."""

tests/units/test_ssh_keys.py

@@ -1,6 +1,6 @@
 """Test the SSH Keys functionality."""
 
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
 
@@ -10,7 +10,9 @@
     InvalidSSHKeyIDError,
 )
 from pytfe.models.ssh_key import (
+    SSHKey,
     SSHKeyCreateOptions,
+    SSHKeyListOptions,
     SSHKeyUpdateOptions,
 )
 from pytfe.resources.ssh_keys import SSHKeys
@@ -51,7 +53,8 @@ def ssh_keys_service(self):
     def test_list_ssh_keys_invalid_org(self, ssh_keys_service):
         """Test listing SSH keys with invalid organization."""
         with pytest.raises(InvalidOrgError):
-            ssh_keys_service.list("")
+            # Need to consume the iterator to trigger the error
+            list(ssh_keys_service.list(""))
 
     def test_create_ssh_key_invalid_org(self, ssh_keys_service):
         """Test creating SSH key with invalid organization."""
@@ -74,3 +77,61 @@ def test_delete_ssh_key_invalid_id(self, ssh_keys_service):
         """Test deleting SSH key with invalid ID."""
         with pytest.raises(InvalidSSHKeyIDError):
             ssh_keys_service.delete("")
+
+    def test_list_ssh_keys_success(self, ssh_keys_service):
+        """Test successful list operation with iterator."""
+        mock_list_data = [
+            {
+                "id": "sshkey-123",
+                "attributes": {
+                    "name": "Test SSH Key 1",
+                },
+            },
+            {
+                "id": "sshkey-456",
+                "attributes": {
+                    "name": "Test SSH Key 2",
+                },
+            },
+        ]
+
+        with patch.object(ssh_keys_service, "_list") as mock_list:
+            mock_list.return_value = mock_list_data
+
+            # Test with options
+            options = SSHKeyListOptions(page_number=1, page_size=5)
+            result = list(ssh_keys_service.list("test-org", options))
+
+            # Verify _list was called with correct path
+            assert mock_list.call_count == 1
+            call_args = mock_list.call_args
+            assert call_args[0][0] == "/api/v2/organizations/test-org/ssh-keys"
+
+            # Verify params structure includes pagination and options
+            params = call_args[1]["params"]
+            assert "page[number]" in params
+            assert "page[size]" in params
+
+            # Verify result structure - iterator yields SSHKey objects
+            assert len(result) == 2
+
+            # Verify SSH key objects were created correctly from response data
+            key1 = result[0]
+            assert isinstance(key1, SSHKey)
+            assert key1.id == "sshkey-123"
+            assert key1.name == "Test SSH Key 1"
+
+            key2 = result[1]
+            assert isinstance(key2, SSHKey)
+            assert key2.id == "sshkey-456"
+            assert key2.name == "Test SSH Key 2"
+
+    def test_list_ssh_keys_empty(self, ssh_keys_service):
+        """Test list operation returning empty results."""
+        with patch.object(ssh_keys_service, "_list") as mock_list:
+            mock_list.return_value = []
+
+            result = list(ssh_keys_service.list("test-org"))
+
+            assert len(result) == 0
+            mock_list.assert_called_once()