Refactor NotificationConfigurations.list() to use iterator pattern

examples/notification_configuration.py

@@ -32,9 +32,15 @@ def main():
 
     print("=== Python TFE Notification Configuration Example ===\n")
 
-    # Resolve workspace and team from environment (fallback to demo placeholders)
-    workspace_id = os.getenv("TFE_WORKSPACE_ID", "ws-example123456789")
-    workspace_name = os.getenv("TFE_ORG", "your-workspace-name")
+    # Resolve organization and workspace from environment variables
+    org_name = os.environ["TFE_ORG"]
+    workspace_name = os.getenv("TFE_WORKSPACE_NAME", "test-api")
+    workspace_id = os.getenv("TFE_WORKSPACE_ID", "")
+    if not workspace_id:
+        print(f"Looking up workspace '{workspace_name}' in org '{org_name}'...")
+        ws = client.workspaces.read(workspace_name, organization=org_name)
+        workspace_id = ws.id
+        print(f"Resolved workspace ID: {workspace_id}")
     print(f"Using workspace: {workspace_name} (ID: {workspace_id})")
 
     team_id = os.getenv("TFE_TEAM_ID", "team-example123456789")
@@ -47,13 +53,12 @@ def main():
         # ===== List notification configurations for workspace =====
         print("1. Listing notification configurations for workspace...")
         try:
-            workspace_notifications = client.notification_configurations.list(
+            workspace_iter = client.notification_configurations.list(
                 subscribable_id=workspace_id
             )
-            print(
-                f"Found {len(workspace_notifications.items)} notification configurations"
-            )
-            for nc in workspace_notifications.items:
+            workspace_notifications = list(workspace_iter)
+            print(f"Found {len(workspace_notifications)} notification configurations")
+            for nc in workspace_notifications:
                 print(f"- {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})")
         except Exception as e:
             print(f"Error listing workspace notifications: {e}")
@@ -69,13 +74,12 @@ def main():
             options = NotificationConfigurationListOptions(
                 subscribable_choice=team_choice
             )
-            team_notifications = client.notification_configurations.list(
+            team_iter = client.notification_configurations.list(
                 subscribable_id=team_id, options=options
             )
-            print(
-                f"Found {len(team_notifications.items)} team notification configurations"
-            )
-            for nc in team_notifications.items:
+            team_notifications = list(team_iter)
+            print(f"Found {len(team_notifications)} team notification configurations")
+            for nc in team_notifications:
                 print(f"- {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})")
         except Exception as e:
             error_msg = str(e).lower()
@@ -93,16 +97,20 @@ def main():
             workspace_choice = NotificationConfigurationSubscribableChoice(
                 workspace={"id": workspace_id}
             )
-            slack_url = os.getenv(
+            # Use GENERIC destination type with a URL that returns HTTP 200.
+            # SLACK/MICROSOFT_TEAMS destinations are auto-verified by HCP Terraform
+            # at creation time; a fake Slack URL returns 302 and causes the create
+            # call to fail immediately.  GENERIC webhooks + httpbin always succeed.
+            webhook_url = os.getenv(
                 "WEBHOOK_URL",
-                "https://hooks.slack.com/services/YOUR_SLACK_WORKSPACE/YOUR_CHANNEL/YOUR_WEBHOOK_TOKEN",
+                "https://httpbin.org/status/200",
             )
             create_options = NotificationConfigurationCreateOptions(
-                destination_type=NotificationDestinationType.SLACK,
+                destination_type=NotificationDestinationType.GENERIC,
                 enabled=True,
-                name="Python TFE Example Slack Notification",
+                name="Python TFE Example Generic Notification",
                 subscribable_choice=workspace_choice,
-                url=slack_url,
+                url=webhook_url,
                 triggers=[
                     NotificationTriggerType.COMPLETED,
                     NotificationTriggerType.ERRORED,
@@ -175,12 +183,17 @@ def main():
 
         except Exception as e:
             error_msg = str(e).lower()
-            if "verification failed" in error_msg and "404" in error_msg:
-                print(" Webhook verification failed (expected with fake URL)")
-                print("The fake Slack URL returns 404 - this is normal for testing")
-                print("To test real verification, use a webhook from:")
-                print("webhook.site (instant test URL)")
-                print("Slack, Teams, or Discord webhook")
+            if "verification failed" in error_msg and (
+                "404" in error_msg or "302" in error_msg
+            ):
+                print("Webhook verification failed (expected with fake URL)")
+                print(
+                    "The URL returned a non-200 response - this is normal for testing"
+                )
+                print("To test real verification, use a webhook from webhook.site,")
+                print(
+                    "Slack, Teams, or Discord, or set WEBHOOK_URL=https://httpbin.org/status/200"
+                )
             else:
                 print(f" Error in workspace notification operations: {e}")
 

src/pytfe/models/notification_configuration.py

@@ -188,26 +188,21 @@ class NotificationConfigurationListOptions:
     """Represents the options for listing notification configurations."""
 
     # Type annotations for instance attributes
-    page_number: int | None
     page_size: int | None
     subscribable_choice: NotificationConfigurationSubscribableChoice | None
 
     def __init__(
         self,
-        page_number: int | None = None,
         page_size: int | None = None,
         subscribable_choice: NotificationConfigurationSubscribableChoice | None = None,
     ):
-        self.page_number = page_number
         self.page_size = page_size
         self.subscribable_choice = subscribable_choice
 
     def to_dict(self) -> dict[str, Any]:
         """Convert to dictionary for API requests."""
         params = {}
 
-        if self.page_number is not None:
-            params["page[number]"] = self.page_number
         if self.page_size is not None:
             params["page[size]"] = self.page_size
 

src/pytfe/resources/notification_configuration.py

@@ -6,6 +6,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterator
 from typing import Any
 
 from ..errors import (
@@ -15,7 +16,6 @@
 from ..models.notification_configuration import (
     NotificationConfiguration,
     NotificationConfigurationCreateOptions,
-    NotificationConfigurationList,
     NotificationConfigurationListOptions,
     NotificationConfigurationUpdateOptions,
 )
@@ -30,35 +30,21 @@ def list(
         self,
         subscribable_id: str,
         options: NotificationConfigurationListOptions | None = None,
-    ) -> NotificationConfigurationList:
+    ) -> Iterator[NotificationConfiguration]:
         """List all notification configurations associated with a workspace or team."""
         if not valid_string_id(subscribable_id):
             raise InvalidOrgError("Invalid subscribable ID")
 
         # Determine URL based on subscribable choice
         if options and options.subscribable_choice and options.subscribable_choice.team:
-            url = f"/api/v2/teams/{subscribable_id}/notification-configurations"
+            path = f"/api/v2/teams/{subscribable_id}/notification-configurations"
         else:
-            url = f"/api/v2/workspaces/{subscribable_id}/notification-configurations"
+            path = f"/api/v2/workspaces/{subscribable_id}/notification-configurations"
 
         params = options.to_dict() if options else None
 
-        r = self.t.request("GET", url, params=params)
-        jd = r.json()
-
-        items = []
-        meta = jd.get("meta", {})
-        pagination = meta.get("pagination", {})
-
-        for d in jd.get("data", []):
-            items.append(self._parse_notification_configuration(d))
-
-        return NotificationConfigurationList(
-            {
-                "data": [{"attributes": item.__dict__} for item in items],
-                "meta": {"pagination": pagination},
-            }
-        )
+        for d in self._list(path, params=params):
+            yield self._parse_notification_configuration(d)
 
     def create(
         self, subscribable_id: str, options: NotificationConfigurationCreateOptions

tests/units/test_notification_configuration.py

@@ -73,20 +73,28 @@ def test_list_workspace_notifications(self):
 
         # Test list operation
         workspace_id = "ws-123456789"
-        result = self.notifications.list(workspace_id)
+        result_iter = self.notifications.list(workspace_id)
+        items = list(result_iter)
 
-        # Verify API call
-        self.mock_transport.request.assert_called_once_with(
-            "GET",
-            f"/api/v2/workspaces/{workspace_id}/notification-configurations",
-            params=None,
+        # Verify API call (occurs when iterator is consumed)
+        self.mock_transport.request.assert_called_once()
+        call_args = self.mock_transport.request.call_args
+        assert call_args[0][0] == "GET"
+        assert (
+            call_args[0][1]
+            == f"/api/v2/workspaces/{workspace_id}/notification-configurations"
         )
+        params = call_args[1].get("params")
+        assert isinstance(params, dict)
+        assert "page[number]" in params and "page[size]" in params
+        assert params["page[number]"] == 1
+        assert params["page[size]"] == 100
 
         # Verify result
-        assert isinstance(result, NotificationConfigurationList)
-        assert len(result.items) == 1
-        assert result.items[0].id == "nc-123456789"
-        assert result.items[0].name == "Test Notification"
+        assert len(items) == 1
+        assert isinstance(items[0], NotificationConfiguration)
+        assert items[0].id == "nc-123456789"
+        assert items[0].name == "Test Notification"
 
     def test_list_team_notifications(self):
         """Test listing notification configurations for a team."""
@@ -105,16 +113,23 @@ def test_list_team_notifications(self):
         team_choice = NotificationConfigurationSubscribableChoice(team={"id": team_id})
         options = NotificationConfigurationListOptions(subscribable_choice=team_choice)
 
-        result = self.notifications.list(team_id, options)
+        result_iter = self.notifications.list(team_id, options)
+        items = list(result_iter)
 
-        # Verify API call
-        self.mock_transport.request.assert_called_once_with(
-            "GET", f"/api/v2/teams/{team_id}/notification-configurations", params={}
-        )
+        # Verify API call (occurs when iterator is consumed)
+        self.mock_transport.request.assert_called_once()
+        call_args = self.mock_transport.request.call_args
+        assert call_args[0][0] == "GET"
+        assert call_args[0][1] == f"/api/v2/teams/{team_id}/notification-configurations"
+        params = call_args[1].get("params")
+        assert isinstance(params, dict)
+        assert "page[number]" in params and "page[size]" in params
+        assert params["page[number]"] == 1
+        assert params["page[size]"] == 100
 
         # Verify result
-        assert isinstance(result, NotificationConfigurationList)
-        assert len(result.items) == 1
+        assert len(items) == 1
+        assert isinstance(items[0], NotificationConfiguration)
 
     def test_list_with_pagination(self):
         """Test listing with pagination options."""
@@ -130,21 +145,29 @@ def test_list_with_pagination(self):
 
         # Test with pagination
         workspace_id = "ws-123456789"
-        options = NotificationConfigurationListOptions(page_number=2, page_size=50)
+        options = NotificationConfigurationListOptions(page_size=50)
 
-        self.notifications.list(workspace_id, options)
+        result_iter = self.notifications.list(workspace_id, options)
+        _ = list(result_iter)
 
-        # Verify API call with pagination
-        self.mock_transport.request.assert_called_once_with(
-            "GET",
-            f"/api/v2/workspaces/{workspace_id}/notification-configurations",
-            params={"page[number]": 2, "page[size]": 50},
+        # page_size from options is respected by _list(); page[number] is controlled by _list()
+        self.mock_transport.request.assert_called_once()
+        call_args = self.mock_transport.request.call_args
+        assert call_args[0][0] == "GET"
+        assert (
+            call_args[0][1]
+            == f"/api/v2/workspaces/{workspace_id}/notification-configurations"
         )
+        params = call_args[1].get("params")
+        assert isinstance(params, dict)
+        assert "page[number]" in params and "page[size]" in params
+        assert params["page[number]"] == 1
+        assert params["page[size]"] == 50
 
     def test_list_invalid_id(self):
         """Test list with invalid subscribable ID."""
         with pytest.raises(InvalidOrgError):
-            self.notifications.list("")
+            list(self.notifications.list(""))
 
     def test_create_workspace_notification(self):
         """Test creating a notification configuration for a workspace."""
@@ -619,10 +642,10 @@ def test_notification_configuration_list(self):
 
     def test_list_options_to_dict(self):
         """Test list options conversion to dictionary."""
-        options = NotificationConfigurationListOptions(page_number=2, page_size=50)
+        options = NotificationConfigurationListOptions(page_size=50)
         result = options.to_dict()
 
-        assert result == {"page[number]": 2, "page[size]": 50}
+        assert result == {"page[size]": 50}
 
     def test_create_options_to_dict(self):
         """Test create options conversion to dictionary."""