diff --git a/examples/organization_tags.py b/examples/organization_tags.py new file mode 100644 index 0000000..c562f57 --- /dev/null +++ b/examples/organization_tags.py @@ -0,0 +1,119 @@ +#!/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 +2. add_workspaces() - associate a workspace with a tag +3. delete() - delete a tag from an organization + +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, +) + + +def main() -> None: + 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 + + 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={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={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 + + 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 + + # 2) Add workspace to tag + if args.workspace_id: + print("[ADD_WORKSPACES] Associating a workspace to a tag") + print( + f"[ADD_WORKSPACES] organization={args.org}, tag_id={args.tag_id}, workspace_id={args.workspace_id}" + ) + try: + client.organization_tags.add_workspaces( + 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}") + else: + print("[ADD_WORKSPACES] skipped: set --workspace-id or TFE_WORKSPACE_ID") + + # 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"[DELETE] API error: {exc}") + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 0e88e1a..4f22961 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -13,6 +13,7 @@ from .resources.oauth_client import OAuthClients from .resources.oauth_token import OAuthTokens from .resources.organization_membership import OrganizationMemberships +from .resources.organization_tags import OrganizationTags from .resources.organization_token import OrganizationTokens from .resources.organizations import Organizations from .resources.plan import Plans @@ -76,6 +77,7 @@ def __init__(self, config: TFEConfig | None = None): self.organizations = Organizations(self._transport) self.organization_memberships = OrganizationMemberships(self._transport) self.users = Users(self._transport) + self.organization_tags = OrganizationTags(self._transport) self.organization_tokens = OrganizationTokens(self._transport) self.projects = Projects(self._transport) self.variables = Variables(self._transport) diff --git a/src/pytfe/errors.py b/src/pytfe/errors.py index f2340af..e9992da 100644 --- a/src/pytfe/errors.py +++ b/src/pytfe/errors.py @@ -121,6 +121,11 @@ class ErrStateVersionUploadNotSupported(TFEError): ... ERR_REQUIRED_TAG_KEY = "tag key is required" ERR_INVALID_TAG_KEY = "invalid tag key" +# Organization Tag Error Constants +ERR_INVALID_TAG = "invalid value for tag" +ERR_REQUIRED_TAG_ID = "tag ID is required" +ERR_REQUIRED_TAG_WORKSPACE_ID = "workspace ID is required" + class WorkspaceNotFound(NotFound): ... diff --git a/src/pytfe/models/organization_tags.py b/src/pytfe/models/organization_tags.py new file mode 100644 index 0000000..1bb1e47 --- /dev/null +++ b/src/pytfe/models/organization_tags.py @@ -0,0 +1,55 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + +from .organization import Organization + + +class OrganizationTag(BaseModel): + """Terraform Enterprise organization tag.""" + + model_config = ConfigDict(populate_by_name=True, extra="forbid") + + id: str = Field(..., description="Tag ID") + name: str | None = Field(None, description="Tag name") + instance_count: int | None = Field( + None, + alias="instance-count", + description="Number of workspaces that have this tag", + ) + organization: Organization | None = Field( + None, + description="Organization this tag belongs to", + ) + + +class OrganizationTagsListOptions(BaseModel): + """Options for listing organization tags.""" + + model_config = ConfigDict(populate_by_name=True, extra="forbid") + + filter: str | None = Field(None, alias="filter[exclude][taggable][id]") + query: str | None = Field( + None, + alias="q", + description="Search query string for tag name likeness", + ) + + +class OrganizationTagsDeleteOptions(BaseModel): + """Options for deleting tags from an organization.""" + + model_config = ConfigDict(extra="forbid") + + ids: list[str] = Field(default_factory=list) + + +class AddWorkspacesToTagOptions(BaseModel): + """Options for associating workspaces with a tag.""" + + model_config = ConfigDict(extra="forbid") + + workspace_ids: list[str] = Field(default_factory=list) diff --git a/src/pytfe/resources/organization_tags.py b/src/pytfe/resources/organization_tags.py new file mode 100644 index 0000000..a092795 --- /dev/null +++ b/src/pytfe/resources/organization_tags.py @@ -0,0 +1,111 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any +from urllib.parse import quote + +from ..errors import ( + ERR_INVALID_ORG, + ERR_INVALID_TAG, + ERR_REQUIRED_TAG_ID, + ERR_REQUIRED_TAG_WORKSPACE_ID, +) +from ..models.organization import Organization +from ..models.organization_tags import ( + AddWorkspacesToTagOptions, + OrganizationTag, + OrganizationTagsDeleteOptions, + OrganizationTagsListOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class OrganizationTags(_Service): + """Organization tags service for Terraform Enterprise.""" + + def list( + self, + organization: str, + options: OrganizationTagsListOptions | None = None, + ) -> 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 {} + for item in self._list(path, params=params): + yield self._parse_organization_tag(item) + + def delete( + self, + organization: str, + options: OrganizationTagsDeleteOptions, + ) -> None: + """Delete tags from an organization.""" + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + if len(options.ids) == 0: + raise ValueError(ERR_REQUIRED_TAG_ID) + + for tag_id in options.ids: + if not valid_string_id(tag_id): + raise ValueError(f"{tag_id} is not a valid id value") + + body = {"data": [{"type": "tags", "id": tag_id} for tag_id in options.ids]} + path = f"/api/v2/organizations/{quote(organization)}/tags" + self.t.request("DELETE", path, json_body=body) + + def add_workspaces( + self, organization: str, tag: str, options: AddWorkspacesToTagOptions + ) -> None: + """Associate workspaces with an organization tag.""" + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + if not valid_string_id(tag): + raise ValueError(ERR_INVALID_TAG) + + if len(options.workspace_ids) == 0: + raise ValueError(ERR_REQUIRED_TAG_WORKSPACE_ID) + + for workspace_id in options.workspace_ids: + if not valid_string_id(workspace_id): + raise ValueError(f"{workspace_id} is not a valid id value") + + body = { + "data": [ + {"type": "workspaces", "id": workspace_id} + for workspace_id in options.workspace_ids + ] + } + path = f"/api/v2/tags/{quote(tag)}/relationships/workspaces" + self.t.request("POST", path, json_body=body) + + def _parse_organization_tag(self, data: dict[str, Any]) -> OrganizationTag: + attributes = data.get("attributes", {}) + relationships = data.get("relationships", {}) + + org = None + org_data = relationships.get("organization", {}).get("data") + if org_data and isinstance(org_data, dict): + org = Organization(id=org_data.get("id")) + + return OrganizationTag.model_validate( + { + "id": data.get("id", ""), + "name": attributes.get("name"), + "instance-count": attributes.get("instance-count"), + "organization": org, + } + ) diff --git a/tests/units/test_organization_tags.py b/tests/units/test_organization_tags.py new file mode 100644 index 0000000..ee23aff --- /dev/null +++ b/tests/units/test_organization_tags.py @@ -0,0 +1,139 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +"""Unit tests for the organization tags module.""" + +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import ( + ERR_INVALID_ORG, + ERR_INVALID_TAG, + ERR_REQUIRED_TAG_ID, + ERR_REQUIRED_TAG_WORKSPACE_ID, +) +from pytfe.models.organization_tags import ( + AddWorkspacesToTagOptions, + OrganizationTagsDeleteOptions, + OrganizationTagsListOptions, +) +from pytfe.resources.organization_tags import OrganizationTags + + +class TestOrganizationTags: + """Test the OrganizationTags service class.""" + + @pytest.fixture + def mock_transport(self): + return Mock(spec=HTTPTransport) + + @pytest.fixture + def organization_tags_service(self, mock_transport): + return OrganizationTags(mock_transport) + + def test_list_success(self, organization_tags_service): + 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 = 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): + organization_tags_service.list("") + + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + organization_tags_service.list(None) + + def test_delete_success(self, organization_tags_service): + with patch.object(organization_tags_service, "t") as mock_t: + mock_t.request.return_value = Mock() + + options = OrganizationTagsDeleteOptions(ids=["tag-1", "tag-2"]) + organization_tags_service.delete("test-org", options) + + call_args = mock_t.request.call_args + assert call_args[0][0] == "DELETE" + assert call_args[0][1] == "/api/v2/organizations/test-org/tags" + assert call_args[1]["json_body"] == { + "data": [ + {"type": "tags", "id": "tag-1"}, + {"type": "tags", "id": "tag-2"}, + ] + } + + def test_delete_validation_errors(self, organization_tags_service): + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + organization_tags_service.delete( + "", OrganizationTagsDeleteOptions(ids=["tag-1"]) + ) + + with pytest.raises(ValueError, match=ERR_REQUIRED_TAG_ID): + organization_tags_service.delete( + "test-org", OrganizationTagsDeleteOptions() + ) + + with pytest.raises(ValueError, match="is not a valid id value"): + organization_tags_service.delete( + "test-org", OrganizationTagsDeleteOptions(ids=[""]) + ) + + def test_add_workspaces_success(self, organization_tags_service): + with patch.object(organization_tags_service, "t") as mock_t: + mock_t.request.return_value = Mock() + + options = AddWorkspacesToTagOptions(workspace_ids=["ws-1", "ws-2"]) + organization_tags_service.add_workspaces("test-org", "tag-1", options) + + call_args = mock_t.request.call_args + assert call_args[0][0] == "POST" + assert call_args[0][1] == "/api/v2/tags/tag-1/relationships/workspaces" + assert call_args[1]["json_body"] == { + "data": [ + {"type": "workspaces", "id": "ws-1"}, + {"type": "workspaces", "id": "ws-2"}, + ] + } + + def test_add_workspaces_validation_errors(self, organization_tags_service): + with pytest.raises(ValueError, match=ERR_INVALID_ORG): + organization_tags_service.add_workspaces( + "", "tag-1", AddWorkspacesToTagOptions(workspace_ids=["ws-1"]) + ) + + with pytest.raises(ValueError, match=ERR_INVALID_TAG): + organization_tags_service.add_workspaces( + "test-org", "", AddWorkspacesToTagOptions(workspace_ids=["ws-1"]) + ) + + with pytest.raises(ValueError, match=ERR_REQUIRED_TAG_WORKSPACE_ID): + organization_tags_service.add_workspaces( + "test-org", "tag-1", AddWorkspacesToTagOptions() + ) + + with pytest.raises(ValueError, match="is not a valid id value"): + organization_tags_service.add_workspaces( + "test-org", "tag-1", AddWorkspacesToTagOptions(workspace_ids=[""]) + )