From 5a05e3bce004b183fdea06ca3022867e8c86d8a7 Mon Sep 17 00:00:00 2001 From: Tanya Singh Date: Mon, 2 Mar 2026 13:31:25 +0530 Subject: [PATCH 1/5] refactor(notification-configuration): convert list to iterator pattern --- examples/notification_configuration.py | 16 ++--- .../resources/notification_configuration.py | 32 ++++----- .../units/test_notification_configuration.py | 69 +++++++++++++------ 3 files changed, 68 insertions(+), 49 deletions(-) diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py index 789367f7..b08d9ff5 100644 --- a/examples/notification_configuration.py +++ b/examples/notification_configuration.py @@ -47,13 +47,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 +68,14 @@ 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 ) + team_notifications = list(team_iter) print( - f"Found {len(team_notifications.items)} team notification configurations" + f"Found {len(team_notifications)} team notification configurations" ) - for nc in team_notifications.items: + for nc in team_notifications: print(f"- {nc.name} (ID: {nc.id}, Enabled: {nc.enabled})") except Exception as e: error_msg = str(e).lower() diff --git a/src/pytfe/resources/notification_configuration.py b/src/pytfe/resources/notification_configuration.py index 4de32ea0..83647004 100644 --- a/src/pytfe/resources/notification_configuration.py +++ b/src/pytfe/resources/notification_configuration.py @@ -6,7 +6,8 @@ from __future__ import annotations -from typing import Any +from typing import Any, Iterator + from ..errors import ( InvalidOrgError, @@ -30,35 +31,28 @@ 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 + if params: + params.pop("page[number]", None) + params.pop("page[size]", None) + params.pop("page_number", 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)) + def _gen(): + for d in self._list(path, params=params): + yield self._parse_notification_configuration(d) - return NotificationConfigurationList( - { - "data": [{"attributes": item.__dict__} for item in items], - "meta": {"pagination": pagination}, - } - ) + return _gen() def create( self, subscribable_id: str, options: NotificationConfigurationCreateOptions diff --git a/tests/units/test_notification_configuration.py b/tests/units/test_notification_configuration.py index 034d161e..05260f8a 100644 --- a/tests/units/test_notification_configuration.py +++ b/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,25 @@ 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.""" @@ -132,14 +149,22 @@ def test_list_with_pagination(self): workspace_id = "ws-123456789" options = NotificationConfigurationListOptions(page_number=2, 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}, + # Verify API call with default pagination keys + 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 def test_list_invalid_id(self): """Test list with invalid subscribable ID.""" From eb667a3cccaf6e96fdf7a8640e0bc75c6df55232 Mon Sep 17 00:00:00 2001 From: Tanya Singh Date: Mon, 2 Mar 2026 13:38:07 +0530 Subject: [PATCH 2/5] chore(examples): update list() usage for iterator pattern --- examples/policy_check.py | 10 +++++----- examples/policy_set.py | 10 +++++----- examples/run_events.py | 13 ++++++------- examples/ssh_keys.py | 18 +++++++++--------- examples/state_versions.py | 28 ++++++++++++++++------------ examples/workspace.py | 6 +++--- 6 files changed, 44 insertions(+), 41 deletions(-) diff --git a/examples/policy_check.py b/examples/policy_check.py index 67f20813..e8515c1d 100644 --- a/examples/policy_check.py +++ b/examples/policy_check.py @@ -55,16 +55,16 @@ def main(): ) try: - pc_list = client.policy_checks.list(args.run_id, options) + pcs = list(client.policy_checks.list(args.run_id, options)) - print(f"Total policy checks: {pc_list.total_count}") - print(f"Page {pc_list.current_page} of {pc_list.total_pages}") + print(f"Total policy checks: {len(pcs)}") + print("Page N/A of N/A") print() - if not pc_list.items: + if not pcs: print("No policy checks found for this run.") else: - for pc in pc_list.items: + for pc in pcs: print(f"- ID: {pc.id}") print(f"Status: {pc.status}") print(f"Scope: {pc.scope}") diff --git a/examples/policy_set.py b/examples/policy_set.py index 1808d80a..b50e7d23 100644 --- a/examples/policy_set.py +++ b/examples/policy_set.py @@ -163,16 +163,16 @@ def main(): ) try: - ps_list = client.policy_sets.list(args.org, list_options) + ps_items = list(client.policy_sets.list(args.org, list_options)) - print(f"Total policy sets: {ps_list.total_count}") - print(f"Page {ps_list.current_page} of {ps_list.total_pages}") + print(f"Total policy sets: {len(ps_items)}") + print("Page N/A of N/A") print() - if not ps_list.items: + if not ps_items: print("No policy sets found for this organization.") else: - for ps in ps_list.items: + for ps in ps_items: print( f"- ID: {ps.id} | Name: {ps.name} | Kind: {ps.kind} | Global: {ps.Global}" ) diff --git a/examples/run_events.py b/examples/run_events.py index a648c5b5..630ff60e 100644 --- a/examples/run_events.py +++ b/examples/run_events.py @@ -94,17 +94,16 @@ def main(): options = RunEventListOptions(include=include_opts if include_opts else None) try: - event_list = client.run_events.list(args.run_id, options) + events = list(client.run_events.list(args.run_id, options)) - print(f"Total run events: {event_list.total_count or 'N/A'}") - if event_list.current_page and event_list.total_pages: - print(f"Page {event_list.current_page} of {event_list.total_pages}") + print(f"Total run events: {len(events)}") + print("Page N/A of N/A") print() - if not event_list.items: + if not events: print("No run events found for this run.") else: - for event in event_list.items: + for event in events: print(f"Event ID: {event.id}") print(f"Action: {event.action or 'N/A'}") print(f"Description: {event.description or 'N/A'}") @@ -139,7 +138,7 @@ def main(): # 3) Summary _print_header("Summary") print(f"Successfully demonstrated run events for run: {args.run_id}") - print(f"Total events found: {event_list.total_count or 'N/A'}") + print(f"Total events found: {len(events)}") if args.event_id: print(f"Successfully read specific event: {args.event_id}") return 0 diff --git a/examples/ssh_keys.py b/examples/ssh_keys.py index 743bd98b..45aa6c0f 100644 --- a/examples/ssh_keys.py +++ b/examples/ssh_keys.py @@ -99,9 +99,9 @@ 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 = list(client.ssh_keys.list(TFE_ORG)) + print(f"Found {len(ssh_keys)} SSH keys:") + for key in ssh_keys: print(f"- ID: {key.id}, Name: {key.name}") # 2. Create a new SSH key @@ -132,16 +132,16 @@ 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 = list(client.ssh_keys.list(TFE_ORG)) + print(f"SSH keys after deletion: {len(ssh_keys_after)}") # 7. Demonstrate pagination with options print("\n7. Demonstrating 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_keys = list(client.ssh_keys.list(TFE_ORG, list_options)) + print(f"Page 1 with page size 5: {len(paginated_keys)} keys") + print("Total pages: N/A") + print(f"Total count: {len(paginated_keys)}") print("\n SSH Keys API example completed successfully!") diff --git a/examples/state_versions.py b/examples/state_versions.py index 6d3f8b82..8bc6cc9b 100644 --- a/examples/state_versions.py +++ b/examples/state_versions.py @@ -47,23 +47,25 @@ def main(): workspace=args.workspace, ) - sv_list = client.state_versions.list(options) + sv_list = list(client.state_versions.list(options)) - print(f"Total state versions: {sv_list.total_count}") - print(f"Page {sv_list.current_page} of {sv_list.total_pages}") + print(f"Total state versions: {len(sv_list)}") + print("Page N/A of N/A") print() - for sv in sv_list.items: + for sv in sv_list: print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") # 1) List all state versions across org and workspace filters _print_header("Org-scoped listing via /api/v2/state-versions (first page)") - all_sv = client.state_versions.list( - StateVersionListOptions( - organization=args.org, workspace=args.workspace, page_size=args.page_size + all_sv = list( + client.state_versions.list( + StateVersionListOptions( + organization=args.org, workspace=args.workspace, page_size=args.page_size + ) ) ) - for sv in all_sv.items: + for sv in all_sv: print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") # 2) Read the current state version (with outputs included if you want) @@ -84,12 +86,14 @@ def main(): # 4) List outputs for the current state version (paged) _print_header("Listing outputs (current state version)") - outs = client.state_versions.list_outputs( - current.id, options=StateVersionOutputsListOptions(page_size=50) + outs = list( + client.state_versions.list_outputs( + current.id, options=StateVersionOutputsListOptions(page_size=50) + ) ) - if not outs.items: + if not outs: print("No outputs found.") - for o in outs.items: + for o in outs: # Sensitive outputs will have value = None print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") diff --git a/examples/workspace.py b/examples/workspace.py index 4dfb6435..6fe125a2 100644 --- a/examples/workspace.py +++ b/examples/workspace.py @@ -435,9 +435,9 @@ def main(): # First, list available SSH keys try: print("Listing available SSH keys...") - ssh_keys = client.ssh_keys.list(args.org) - if ssh_keys.items: - ssh_key = ssh_keys.items[0] + ssh_keys = list(client.ssh_keys.list(args.org)) + if ssh_keys: + ssh_key = ssh_keys[0] print(f"Found SSH key: {ssh_key.name} (ID: {ssh_key.id})") # Test assign SSH key From 6c746ba520c6c896088098de5d34ab76506e0e67 Mon Sep 17 00:00:00 2001 From: Tanya Singh Date: Thu, 5 Mar 2026 13:05:15 +0530 Subject: [PATCH 3/5] Finish notification configuration iterator changes --- examples/notification_configuration.py | 47 ++++++++++++------- examples/policy_set.py | 10 ++-- examples/run_events.py | 13 ++--- examples/ssh_keys.py | 18 +++---- examples/state_versions.py | 28 +++++------ examples/variables.py | 11 ++++- examples/workspace.py | 6 +-- .../resources/notification_configuration.py | 7 ++- .../units/test_notification_configuration.py | 4 +- 9 files changed, 79 insertions(+), 65 deletions(-) diff --git a/examples/notification_configuration.py b/examples/notification_configuration.py index b08d9ff5..8086d3e6 100644 --- a/examples/notification_configuration.py +++ b/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") @@ -72,9 +78,7 @@ def main(): subscribable_id=team_id, options=options ) team_notifications = list(team_iter) - print( - f"Found {len(team_notifications)} team notification configurations" - ) + 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: @@ -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}") diff --git a/examples/policy_set.py b/examples/policy_set.py index b50e7d23..1808d80a 100644 --- a/examples/policy_set.py +++ b/examples/policy_set.py @@ -163,16 +163,16 @@ def main(): ) try: - ps_items = list(client.policy_sets.list(args.org, list_options)) + ps_list = client.policy_sets.list(args.org, list_options) - print(f"Total policy sets: {len(ps_items)}") - print("Page N/A of N/A") + print(f"Total policy sets: {ps_list.total_count}") + print(f"Page {ps_list.current_page} of {ps_list.total_pages}") print() - if not ps_items: + if not ps_list.items: print("No policy sets found for this organization.") else: - for ps in ps_items: + for ps in ps_list.items: print( f"- ID: {ps.id} | Name: {ps.name} | Kind: {ps.kind} | Global: {ps.Global}" ) diff --git a/examples/run_events.py b/examples/run_events.py index 630ff60e..a648c5b5 100644 --- a/examples/run_events.py +++ b/examples/run_events.py @@ -94,16 +94,17 @@ def main(): options = RunEventListOptions(include=include_opts if include_opts else None) try: - events = list(client.run_events.list(args.run_id, options)) + event_list = client.run_events.list(args.run_id, options) - print(f"Total run events: {len(events)}") - print("Page N/A of N/A") + print(f"Total run events: {event_list.total_count or 'N/A'}") + if event_list.current_page and event_list.total_pages: + print(f"Page {event_list.current_page} of {event_list.total_pages}") print() - if not events: + if not event_list.items: print("No run events found for this run.") else: - for event in events: + for event in event_list.items: print(f"Event ID: {event.id}") print(f"Action: {event.action or 'N/A'}") print(f"Description: {event.description or 'N/A'}") @@ -138,7 +139,7 @@ def main(): # 3) Summary _print_header("Summary") print(f"Successfully demonstrated run events for run: {args.run_id}") - print(f"Total events found: {len(events)}") + print(f"Total events found: {event_list.total_count or 'N/A'}") if args.event_id: print(f"Successfully read specific event: {args.event_id}") return 0 diff --git a/examples/ssh_keys.py b/examples/ssh_keys.py index 45aa6c0f..743bd98b 100644 --- a/examples/ssh_keys.py +++ b/examples/ssh_keys.py @@ -99,9 +99,9 @@ def main(): try: # 1. List existing SSH keys print("\n1. Listing SSH keys...") - ssh_keys = list(client.ssh_keys.list(TFE_ORG)) - print(f"Found {len(ssh_keys)} SSH keys:") - for key in 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: print(f"- ID: {key.id}, Name: {key.name}") # 2. Create a new SSH key @@ -132,16 +132,16 @@ def main(): # 6. Verify deletion by listing again print("\n6. Verifying deletion...") - ssh_keys_after = list(client.ssh_keys.list(TFE_ORG)) - print(f"SSH keys after deletion: {len(ssh_keys_after)}") + ssh_keys_after = client.ssh_keys.list(TFE_ORG) + print(f"SSH keys after deletion: {len(ssh_keys_after.items)}") # 7. Demonstrate pagination with options print("\n7. Demonstrating pagination options...") list_options = SSHKeyListOptions(page_size=5, page_number=1) - paginated_keys = list(client.ssh_keys.list(TFE_ORG, list_options)) - print(f"Page 1 with page size 5: {len(paginated_keys)} keys") - print("Total pages: N/A") - print(f"Total count: {len(paginated_keys)}") + 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}") print("\n SSH Keys API example completed successfully!") diff --git a/examples/state_versions.py b/examples/state_versions.py index 8bc6cc9b..6d3f8b82 100644 --- a/examples/state_versions.py +++ b/examples/state_versions.py @@ -47,25 +47,23 @@ def main(): workspace=args.workspace, ) - sv_list = list(client.state_versions.list(options)) + sv_list = client.state_versions.list(options) - print(f"Total state versions: {len(sv_list)}") - print("Page N/A of N/A") + print(f"Total state versions: {sv_list.total_count}") + print(f"Page {sv_list.current_page} of {sv_list.total_pages}") print() - for sv in sv_list: + for sv in sv_list.items: print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") # 1) List all state versions across org and workspace filters _print_header("Org-scoped listing via /api/v2/state-versions (first page)") - all_sv = list( - client.state_versions.list( - StateVersionListOptions( - organization=args.org, workspace=args.workspace, page_size=args.page_size - ) + all_sv = client.state_versions.list( + StateVersionListOptions( + organization=args.org, workspace=args.workspace, page_size=args.page_size ) ) - for sv in all_sv: + for sv in all_sv.items: print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") # 2) Read the current state version (with outputs included if you want) @@ -86,14 +84,12 @@ def main(): # 4) List outputs for the current state version (paged) _print_header("Listing outputs (current state version)") - outs = list( - client.state_versions.list_outputs( - current.id, options=StateVersionOutputsListOptions(page_size=50) - ) + outs = client.state_versions.list_outputs( + current.id, options=StateVersionOutputsListOptions(page_size=50) ) - if not outs: + if not outs.items: print("No outputs found.") - for o in outs: + for o in outs.items: # Sensitive outputs will have value = None print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") diff --git a/examples/variables.py b/examples/variables.py index 0b3fe345..6003593b 100644 --- a/examples/variables.py +++ b/examples/variables.py @@ -21,8 +21,15 @@ def main(): # Initialize the TFE client client = TFEClient(TFEConfig.from_env()) - # Replace this with your actual workspace ID - workspace_id = "ws-example123456789" # Get this from your TFE workspace + # 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", "").strip() + 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"Testing all variable operations in workspace: {workspace_id}") print("=" * 60) diff --git a/examples/workspace.py b/examples/workspace.py index 6fe125a2..4dfb6435 100644 --- a/examples/workspace.py +++ b/examples/workspace.py @@ -435,9 +435,9 @@ def main(): # First, list available SSH keys try: print("Listing available SSH keys...") - ssh_keys = list(client.ssh_keys.list(args.org)) - if ssh_keys: - ssh_key = ssh_keys[0] + ssh_keys = client.ssh_keys.list(args.org) + if ssh_keys.items: + ssh_key = ssh_keys.items[0] print(f"Found SSH key: {ssh_key.name} (ID: {ssh_key.id})") # Test assign SSH key diff --git a/src/pytfe/resources/notification_configuration.py b/src/pytfe/resources/notification_configuration.py index 83647004..7d3b731c 100644 --- a/src/pytfe/resources/notification_configuration.py +++ b/src/pytfe/resources/notification_configuration.py @@ -6,8 +6,8 @@ from __future__ import annotations -from typing import Any, Iterator - +from collections.abc import Iterator +from typing import Any from ..errors import ( InvalidOrgError, @@ -16,7 +16,6 @@ from ..models.notification_configuration import ( NotificationConfiguration, NotificationConfigurationCreateOptions, - NotificationConfigurationList, NotificationConfigurationListOptions, NotificationConfigurationUpdateOptions, ) @@ -48,7 +47,7 @@ def list( params.pop("page[size]", None) params.pop("page_number", None) - def _gen(): + def _gen() -> Iterator[NotificationConfiguration]: for d in self._list(path, params=params): yield self._parse_notification_configuration(d) diff --git a/tests/units/test_notification_configuration.py b/tests/units/test_notification_configuration.py index 05260f8a..5b367519 100644 --- a/tests/units/test_notification_configuration.py +++ b/tests/units/test_notification_configuration.py @@ -120,9 +120,7 @@ def test_list_team_notifications(self): 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" - ) + 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 From fe107bd9512efa121e92e01fd1eb5f8b1e4737e7 Mon Sep 17 00:00:00 2001 From: Tanya Singh Date: Tue, 10 Mar 2026 11:26:46 +0530 Subject: [PATCH 4/5] Simplify notification configuration iterator and remove page_number option --- src/pytfe/models/notification_configuration.py | 5 ----- src/pytfe/resources/notification_configuration.py | 11 ++--------- tests/units/test_notification_configuration.py | 12 ++++++------ 3 files changed, 8 insertions(+), 20 deletions(-) diff --git a/src/pytfe/models/notification_configuration.py b/src/pytfe/models/notification_configuration.py index 0632a1e8..2e22d486 100644 --- a/src/pytfe/models/notification_configuration.py +++ b/src/pytfe/models/notification_configuration.py @@ -188,17 +188,14 @@ 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 @@ -206,8 +203,6 @@ 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 diff --git a/src/pytfe/resources/notification_configuration.py b/src/pytfe/resources/notification_configuration.py index 7d3b731c..17341c76 100644 --- a/src/pytfe/resources/notification_configuration.py +++ b/src/pytfe/resources/notification_configuration.py @@ -42,16 +42,9 @@ def list( path = f"/api/v2/workspaces/{subscribable_id}/notification-configurations" params = options.to_dict() if options else None - if params: - params.pop("page[number]", None) - params.pop("page[size]", None) - params.pop("page_number", None) - def _gen() -> Iterator[NotificationConfiguration]: - for d in self._list(path, params=params): - yield self._parse_notification_configuration(d) - - return _gen() + for d in self._list(path, params=params): + yield self._parse_notification_configuration(d) def create( self, subscribable_id: str, options: NotificationConfigurationCreateOptions diff --git a/tests/units/test_notification_configuration.py b/tests/units/test_notification_configuration.py index 5b367519..3da8de97 100644 --- a/tests/units/test_notification_configuration.py +++ b/tests/units/test_notification_configuration.py @@ -145,12 +145,12 @@ 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) result_iter = self.notifications.list(workspace_id, options) _ = list(result_iter) - # Verify API call with default pagination keys + # 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" @@ -162,12 +162,12 @@ def test_list_with_pagination(self): assert isinstance(params, dict) assert "page[number]" in params and "page[size]" in params assert params["page[number]"] == 1 - assert params["page[size]"] == 100 + 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.""" @@ -642,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.""" From c2b1341b19e44b2ea43b2786a810576095a0b614 Mon Sep 17 00:00:00 2001 From: Tanya Singh Date: Thu, 12 Mar 2026 11:18:43 +0530 Subject: [PATCH 5/5] Remove unrelated example changes from notification configuration --- examples/policy_check.py | 10 +++++----- examples/variables.py | 11 ++--------- 2 files changed, 7 insertions(+), 14 deletions(-) diff --git a/examples/policy_check.py b/examples/policy_check.py index e8515c1d..67f20813 100644 --- a/examples/policy_check.py +++ b/examples/policy_check.py @@ -55,16 +55,16 @@ def main(): ) try: - pcs = list(client.policy_checks.list(args.run_id, options)) + pc_list = client.policy_checks.list(args.run_id, options) - print(f"Total policy checks: {len(pcs)}") - print("Page N/A of N/A") + print(f"Total policy checks: {pc_list.total_count}") + print(f"Page {pc_list.current_page} of {pc_list.total_pages}") print() - if not pcs: + if not pc_list.items: print("No policy checks found for this run.") else: - for pc in pcs: + for pc in pc_list.items: print(f"- ID: {pc.id}") print(f"Status: {pc.status}") print(f"Scope: {pc.scope}") diff --git a/examples/variables.py b/examples/variables.py index 6003593b..0b3fe345 100644 --- a/examples/variables.py +++ b/examples/variables.py @@ -21,15 +21,8 @@ def main(): # Initialize the TFE client client = TFEClient(TFEConfig.from_env()) - # 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", "").strip() - 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}") + # Replace this with your actual workspace ID + workspace_id = "ws-example123456789" # Get this from your TFE workspace print(f"Testing all variable operations in workspace: {workspace_id}") print("=" * 60)