refactor organization tags list API to iterator pattern

NimishaShrivastava-dev · NimishaShrivastava-dev · commit 00afd13da0b6 · 2026-05-14T12:46:16.000+05:30
diff --git a/examples/organization_tags.py b/examples/organization_tags.py
@@ -1,82 +1,113 @@
 #!/usr/bin/env python3
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
 """Organization tags operations example.
 
 Demonstrates:
-1. list() - list tags in an organization
+1. list()           - list tags in an organization
+2. add_workspaces() - associate a workspace with a tag
+3. delete()         - delete a tag from an organization
 
-This phase intentionally uses only organization-level parameters.
-Tag IDs and workspace IDs can be passed in a later phase.
+Usage:
+    python examples/organization_tags.py --org my-org
+    python examples/organization_tags.py --org my-org --tag-id tag-abc123 --workspace-id ws-xyz
 """
 
+from __future__ import annotations
+
+import argparse
 import os
 
 from pytfe import TFEClient, TFEConfig
 from pytfe.errors import TFEError
-from pytfe.models.organization_tags import (
-    AddWorkspacesToTagOptions,
-    OrganizationTagsDeleteOptions,
-)
+from pytfe.models.organization_tags import AddWorkspacesToTagOptions, OrganizationTagsDeleteOptions
 
 
 def main() -> None:
-    client = TFEClient(TFEConfig.from_env())
+    parser = argparse.ArgumentParser(description="Organization Tags demo for python-tfe SDK")
+    parser.add_argument(
+        "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io")
+    )
+    parser.add_argument("--token", default=os.getenv("TFE_TOKEN", ""))
+    parser.add_argument(
+        "--org",
+        default=os.getenv("TFE_ORG", ""),
+        help="Organization name",
+    )
+    parser.add_argument(
+        "--tag-id",
+        default=os.getenv("TFE_TAG_ID", ""),
+        help="Tag ID for add/delete operations",
+    )
+    parser.add_argument(
+        "--workspace-id",
+        default=os.getenv("TFE_WORKSPACE_ID", ""),
+        help="Workspace ID to associate with tag",
+    )
+    args = parser.parse_args()
+
+    if not args.token:
+        print("Error: TFE_TOKEN environment variable or --token required")
+        return
 
-    organization_name = os.getenv("TFE_ORG", "example-org")
-    tag_id = os.getenv("TFE_TAG_ID", "")
-    workspace_id = os.getenv("TFE_WORKSPACE_ID", "")
-    operation = "list"
+    if not args.org:
+        print("Error: TFE_ORG environment variable or --org required")
+        return
 
+    cfg = TFEConfig(address=args.address, token=args.token)
+    client = TFEClient(cfg)
+
+    # 1) List tags
     try:
         print("[LIST] Listing organization tags")
-        print(f"[LIST] organization={organization_name}")
-        tags = client.organization_tags.list(organization_name)
-        print(f"[LIST] total_tags={len(tags.items)}")
-        for item in tags.items:
+        print(f"[LIST] organization={args.org}")
+        tags = list(client.organization_tags.list(args.org))
+        print(f"[LIST] total_tags={len(tags)}")
+        for tag in tags:
             print(
-                f"[LIST] id={item.id}, name={item.name}, instance_count={item.instance_count}"
+                f"[LIST] id={tag.id}, name={tag.name}, instance_count={tag.instance_count}"
             )
+        if not tags:
+            print("[LIST] no tags found")
+    except TFEError as exc:
+        print(f"[LIST] API error: {exc}")
+        return
 
-        # Guard: ensure env vars are set
-        if not tag_id or not workspace_id:
-            print("Skipping add/delete: set TFE_TAG_ID and TFE_WORKSPACE_ID first.")
-            return
+    if not args.tag_id:
+        print("[ADD_WORKSPACES] skipped: set --tag-id or TFE_TAG_ID")
+        print("[DELETE] skipped: set --tag-id or TFE_TAG_ID")
+        return
 
-        # ---- Add workspace ----
-        operation = "add_workspaces"
+    # 2) Add workspace to tag
+    if args.workspace_id:
         print("[ADD_WORKSPACES] Associating a workspace to a tag")
         print(
-            f"[ADD_WORKSPACES] organization={organization_name}, tag_id={tag_id}, workspace_id={workspace_id}"
+            f"[ADD_WORKSPACES] organization={args.org}, tag_id={args.tag_id}, workspace_id={args.workspace_id}"
         )
         try:
             client.organization_tags.add_workspaces(
-                organization_name,
-                tag_id,
-                AddWorkspacesToTagOptions(workspace_ids=[workspace_id]),
+                args.org,
+                args.tag_id,
+                AddWorkspacesToTagOptions(workspace_ids=[args.workspace_id]),
             )
             print("[ADD_WORKSPACES] workspace associated")
         except TFEError as exc:
             print(f"[ADD_WORKSPACES] API error: {exc}")
-            print(f"[ADD_WORKSPACES] failed operation={operation}")
+    else:
+        print("[ADD_WORKSPACES] skipped: set --workspace-id or TFE_WORKSPACE_ID")
 
-        # ---- Delete tag ----
-        operation = "delete"
-        print("[DELETE] Deleting a tag from the organization")
-        print(f"[DELETE] organization={organization_name}, tag_id={tag_id}")
-        try:
-            client.organization_tags.delete(
-                organization_name,
-                OrganizationTagsDeleteOptions(ids=[tag_id]),
-            )
-            print("[DELETE] tag deleted")
-        except TFEError as exc:
-            print(f"[DELETE] API error: {exc}")
-            print(f"[DELETE] failed operation={operation}")
+    # 3) Delete tag
+    print("[DELETE] Deleting a tag from the organization")
+    print(f"[DELETE] organization={args.org}, tag_id={args.tag_id}")
+    try:
+        client.organization_tags.delete(
+            args.org,
+            OrganizationTagsDeleteOptions(ids=[args.tag_id]),
+        )
+        print("[DELETE] tag deleted")
     except TFEError as exc:
-        print(f"API error: {exc}")
-        print(f"Failed during operation: {operation}")
-        print("Check TFE_TOKEN, TFE_ADDRESS, and organization/tag/workspace IDs.")
-    finally:
-        client.close()
+        print(f"[DELETE] API error: {exc}")
 
 
 if __name__ == "__main__":
diff --git a/src/pytfe/models/organization_tags.py b/src/pytfe/models/organization_tags.py
@@ -5,7 +5,6 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from .common import Pagination
 from .organization import Organization
 
 
@@ -27,15 +26,6 @@ class OrganizationTag(BaseModel):
     )
 
 
-class OrganizationTagsList(BaseModel):
-    """Represents a list response for organization tags."""
-
-    model_config = ConfigDict(extra="forbid")
-
-    pagination: Pagination | None = Field(None)
-    items: list[OrganizationTag] = Field(default_factory=list)
-
-
 class OrganizationTagsListOptions(BaseModel):
     """Options for listing organization tags."""
 
diff --git a/src/pytfe/resources/organization_tags.py b/src/pytfe/resources/organization_tags.py
@@ -3,19 +3,18 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterator
 from typing import Any
 from urllib.parse import quote
 
 from ..errors import (
     ERR_INVALID_ORG,
 )
-from ..models.common import Pagination
 from ..models.organization import Organization
 from ..models.organization_tags import (
     AddWorkspacesToTagOptions,
     OrganizationTag,
     OrganizationTagsDeleteOptions,
-    OrganizationTagsList,
     OrganizationTagsListOptions,
 )
 from ..utils import valid_string_id
@@ -33,34 +32,21 @@ def list(
         self,
         organization: str,
         options: OrganizationTagsListOptions | None = None,
-    ) -> OrganizationTagsList:
+    ) -> Iterator[OrganizationTag]:
         """List all tags within an organization."""
         if not valid_string_id(organization):
             raise ValueError(ERR_INVALID_ORG)
+        return self._iter_tags(organization, options)
 
+    def _iter_tags(
+        self,
+        organization: str,
+        options: OrganizationTagsListOptions | None = None,
+    ) -> Iterator[OrganizationTag]:
         path = f"/api/v2/organizations/{quote(organization)}/tags"
-        params = (
-            options.model_dump(by_alias=True, exclude_none=True) if options else None
-        )
-
-        response = self.t.request("GET", path, params=params)
-        payload = response.json() or {}
-
-        items = [self._parse_organization_tag(item) for item in payload.get("data", [])]
-
-        pagination = None
-        meta = payload.get("meta", {})
-        pagination_data = meta.get("pagination", {}) if isinstance(meta, dict) else {}
-        if pagination_data:
-            pagination = Pagination(
-                current_page=pagination_data.get("current-page", 1),
-                total_count=pagination_data.get("total-count", len(items)),
-                previous_page=pagination_data.get("previous-page"),
-                next_page=pagination_data.get("next-page"),
-                total_pages=pagination_data.get("total-pages"),
-            )
-
-        return OrganizationTagsList(pagination=pagination, items=items)
+        params = options.model_dump(by_alias=True, exclude_none=True) if options else {}
+        for item in self._list(path, params=params):
+            yield self._parse_organization_tag(item)
 
     def delete(
         self,
diff --git a/tests/units/test_organization_tags.py b/tests/units/test_organization_tags.py
@@ -1,21 +1,19 @@
+# Copyright IBM Corp. 2025, 2026
+# SPDX-License-Identifier: MPL-2.0
+
 """Unit tests for the organization tags module."""
 
-import os
-import sys
 from unittest.mock import Mock, patch
 
 import pytest
 
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "src"))
-
 from pytfe._http import HTTPTransport
 from pytfe.errors import (
     ERR_INVALID_ORG,
 )
 from pytfe.models.organization_tags import (
     AddWorkspacesToTagOptions,
     OrganizationTagsDeleteOptions,
-    OrganizationTagsList,
     OrganizationTagsListOptions,
 )
 from pytfe.resources.organization_tags import OrganizationTags
@@ -37,56 +35,31 @@ def organization_tags_service(self, mock_transport):
         return OrganizationTags(mock_transport)
 
     def test_list_success(self, organization_tags_service):
-        mock_response_data = {
-            "data": [
-                {
-                    "id": "tag-1",
-                    "attributes": {
-                        "name": "env:dev",
-                        "instance-count": 2,
-                    },
-                    "relationships": {
-                        "organization": {
-                            "data": {"id": "org-1", "type": "organizations"}
-                        }
-                    },
-                }
-            ],
-            "meta": {
-                "pagination": {
-                    "current-page": 1,
-                    "total-count": 1,
-                    "next-page": None,
-                    "previous-page": None,
-                    "total-pages": 1,
-                }
-            },
-        }
-
-        mock_response = Mock()
-        mock_response.json.return_value = mock_response_data
-
-        with patch.object(organization_tags_service, "t") as mock_t:
-            mock_t.request.return_value = mock_response
+        mock_items = [
+            {
+                "id": "tag-1",
+                "attributes": {
+                    "name": "env:dev",
+                    "instance-count": 2,
+                },
+                "relationships": {
+                    "organization": {
+                        "data": {"id": "org-1", "type": "organizations"}
+                    }
+                },
+            }
+        ]
 
+        with patch.object(organization_tags_service, "_list", return_value=iter(mock_items)):
             options = OrganizationTagsListOptions(query="env")
-            result = organization_tags_service.list("test-org", options)
-
-            assert isinstance(result, OrganizationTagsList)
-            assert len(result.items) == 1
-            assert result.items[0].id == "tag-1"
-            assert result.items[0].name == "env:dev"
-            assert result.items[0].instance_count == 2
-            assert result.items[0].organization is not None
-            assert result.items[0].organization.id == "org-1"
-            assert result.pagination is not None
-            assert result.pagination.current_page == 1
-            assert result.pagination.total_count == 1
-
-            call_args = mock_t.request.call_args
-            assert call_args[0][0] == "GET"
-            assert call_args[0][1] == "/api/v2/organizations/test-org/tags"
-            assert call_args[1]["params"]["q"] == "env"
+            result = list(organization_tags_service.list("test-org", options))
+
+            assert len(result) == 1
+            assert result[0].id == "tag-1"
+            assert result[0].name == "env:dev"
+            assert result[0].instance_count == 2
+            assert result[0].organization is not None
+            assert result[0].organization.id == "org-1"
 
     def test_list_validation_errors(self, organization_tags_service):
         with pytest.raises(ValueError, match=ERR_INVALID_ORG):
