diff --git a/examples/reserved_tag_key.py b/examples/reserved_tag_key.py new file mode 100644 index 0000000..1c0d1ea --- /dev/null +++ b/examples/reserved_tag_key.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""Reserved Tag Keys Example Script. + +This script demonstrates how to use the Reserved Tag Keys API to: +1. List all reserved tag keys for an organization +2. Create a new reserved tag key +3. Update a reserved tag key +4. Delete a reserved tag key + +Before running this script: +1. Set TFE_TOKEN environment variable with your Terraform Cloud API token +2. Set TFE_ORG environment variable with your organization name +""" + +import os +import sys + +# Add the source directory to the path for direct execution +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from tfe import TFEClient, TFEConfig +from tfe.errors import TFEError +from tfe.models import ( + ReservedTagKeyCreateOptions, + ReservedTagKeyListOptions, + ReservedTagKeyUpdateOptions, +) + +# Configuration +TFE_TOKEN = os.getenv("TFE_TOKEN") +TFE_ORG = os.getenv("TFE_ORG") + + +def main(): + """Main function demonstrating Reserved Tag Keys API usage.""" + + # Validate environment variables + if not TFE_TOKEN: + print("❌ Error: TFE_TOKEN environment variable is required") + sys.exit(1) + + if not TFE_ORG: + print("❌ Error: TFE_ORG environment variable is required") + sys.exit(1) + + # Initialize the TFE client + config = TFEConfig(token=TFE_TOKEN) + client = TFEClient(config) + + print(f"Reserved Tag Keys API Example for organization: {TFE_ORG}") + print("=" * 60) + + try: + # 1. List existing reserved tag keys + print("\n1. Listing reserved tag keys...") + reserved_tag_keys = client.reserved_tag_key.list(TFE_ORG) + print(f"✅ Found {len(reserved_tag_keys.items)} reserved tag keys:") + for rtk in reserved_tag_keys.items: + print( + f" - ID: {rtk.id}, Key: {rtk.key}, Disable Overrides: {rtk.disable_overrides}" + ) + + # 2. Create a new reserved tag key + print("\n2. Creating a new reserved tag key...") + create_options = ReservedTagKeyCreateOptions( + key="python-tfe-example", disable_overrides=False + ) + + new_rtk = client.reserved_tag_key.create(TFE_ORG, create_options) + print(f"✅ Created reserved tag key: {new_rtk.id} - {new_rtk.key}") + print(f" Disable Overrides: {new_rtk.disable_overrides}") + + # 3. Update the reserved tag key + print("\n3. Updating the reserved tag key...") + update_options = ReservedTagKeyUpdateOptions( + key="python-tfe-example-updated", disable_overrides=True + ) + + updated_rtk = client.reserved_tag_key.update(new_rtk.id, update_options) + print(f"✅ Updated reserved tag key: {updated_rtk.id} - {updated_rtk.key}") + print(f" Disable Overrides: {updated_rtk.disable_overrides}") + + # 4. Delete the reserved tag key + print("\n4. Deleting the reserved tag key...") + client.reserved_tag_key.delete(new_rtk.id) + print(f"✅ Deleted reserved tag key: {new_rtk.id}") + + # 5. Verify deletion by listing again + print("\n5. Verifying deletion...") + reserved_tag_keys_after = client.reserved_tag_key.list(TFE_ORG) + print( + f"✅ Reserved tag keys after deletion: {len(reserved_tag_keys_after.items)}" + ) + + # 6. Demonstrate pagination with options + print("\n6. Demonstrating pagination options...") + list_options = ReservedTagKeyListOptions(page_size=5, page_number=1) + paginated_rtks = client.reserved_tag_key.list(TFE_ORG, list_options) + print(f"✅ Page 1 with page size 5: {len(paginated_rtks.items)} keys") + print(f" Total pages: {paginated_rtks.total_pages}") + print(f" Total count: {paginated_rtks.total_count}") + + print("\n🎉 Reserved Tag Keys API example completed successfully!") + + except NotImplementedError as e: + print(f"\n⚠️ Note: {e}") + print("This is expected - the read operation is not supported by the API.") + + except TFEError as e: + print(f"\n❌ TFE API Error: {e}") + if hasattr(e, "status"): + if e.status == 403: + print("💡 Permission denied - check token permissions") + elif e.status == 401: + print("💡 Authentication failed - check token validity") + elif e.status == 422: + print("💡 Validation error - check reserved tag key format") + sys.exit(1) + + except Exception as e: + print(f"\n❌ Unexpected error: {e}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/src/tfe/client.py b/src/tfe/client.py index cd4a402..fa3587b 100644 --- a/src/tfe/client.py +++ b/src/tfe/client.py @@ -13,6 +13,7 @@ from .resources.query_run import QueryRuns from .resources.registry_module import RegistryModules from .resources.registry_provider import RegistryProviders +from .resources.reserved_tag_key import ReservedTagKey from .resources.run import Runs from .resources.run_event import RunEvents from .resources.run_task import RunTasks @@ -73,5 +74,8 @@ def __init__(self, config: TFEConfig | None = None): # SSH Keys self.ssh_keys = SSHKeys(self._transport) + # Reserved Tag Key + self.reserved_tag_key = ReservedTagKey(self._transport) + def close(self) -> None: pass diff --git a/src/tfe/errors.py b/src/tfe/errors.py index 341150f..c633e0f 100644 --- a/src/tfe/errors.py +++ b/src/tfe/errors.py @@ -102,6 +102,11 @@ class ErrStateVersionUploadNotSupported(TFEError): ... # SSH Key Error Constants ERR_INVALID_SSH_KEY_ID = "invalid SSH key ID" +# Reserved Tag Key Error Constants +ERR_INVALID_RESERVED_TAG_KEY_ID = "invalid reserved tag key ID" +ERR_REQUIRED_TAG_KEY = "tag key is required" +ERR_INVALID_TAG_KEY = "invalid tag key" + class WorkspaceNotFound(NotFound): ... diff --git a/src/tfe/models/__init__.py b/src/tfe/models/__init__.py index 1ba6309..24d297d 100644 --- a/src/tfe/models/__init__.py +++ b/src/tfe/models/__init__.py @@ -111,6 +111,15 @@ RegistryProviderReadOptions, ) +# Re-export all reserved tag key types +from .reserved_tag_key import ( + ReservedTagKey, + ReservedTagKeyCreateOptions, + ReservedTagKeyList, + ReservedTagKeyListOptions, + ReservedTagKeyUpdateOptions, +) + # Re-export all SSH key types from .ssh_key import ( SSHKey, @@ -139,6 +148,12 @@ "SSHKeyList", "SSHKeyListOptions", "SSHKeyUpdateOptions", + # Reserved tag key types + "ReservedTagKey", + "ReservedTagKeyCreateOptions", + "ReservedTagKeyList", + "ReservedTagKeyListOptions", + "ReservedTagKeyUpdateOptions", # Agent and agent pool types "Agent", "AgentPool", diff --git a/src/tfe/models/reserved_tag_key.py b/src/tfe/models/reserved_tag_key.py new file mode 100644 index 0000000..eb125ea --- /dev/null +++ b/src/tfe/models/reserved_tag_key.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + + +class ReservedTagKey(BaseModel): + """Represents a reserved tag key in Terraform Enterprise.""" + + model_config = ConfigDict(populate_by_name=True) + + id: str = Field(..., description="The unique identifier for this reserved tag key") + type: str = Field( + default="reserved-tag-keys", description="The type of this resource" + ) + key: str = Field(..., description="The key targeted by this reserved tag key") + disable_overrides: bool = Field( + ..., + alias="disable-overrides", + description="If true, disables overriding inherited tags with the specified key at the workspace level", + ) + created_at: datetime | None = Field( + None, + alias="created-at", + description="The time when the reserved tag key was created", + ) + updated_at: datetime | None = Field( + None, + alias="updated-at", + description="The time when the reserved tag key was last updated", + ) + + +class ReservedTagKeyCreateOptions(BaseModel): + """Options for creating a new reserved tag key.""" + + model_config = ConfigDict(populate_by_name=True) + + key: str = Field(..., description="The key targeted by this reserved tag key") + disable_overrides: bool = Field( + ..., + alias="disable-overrides", + description="If true, disables overriding inherited tags with the specified key at the workspace level", + ) + + +class ReservedTagKeyUpdateOptions(BaseModel): + """Options for updating a reserved tag key.""" + + model_config = ConfigDict(populate_by_name=True) + + key: str | None = Field( + None, description="The key targeted by this reserved tag key" + ) + disable_overrides: bool | None = Field( + None, + alias="disable-overrides", + description="If true, disables overriding inherited tags with the specified key at the workspace level", + ) + + +class ReservedTagKeyListOptions(BaseModel): + """Options for listing reserved tag keys.""" + + model_config = ConfigDict(populate_by_name=True) + + page_number: int | None = Field( + None, alias="page[number]", description="Page number to retrieve", ge=1 + ) + page_size: int | None = Field( + None, alias="page[size]", description="Number of items per page", ge=1, le=100 + ) + + +class ReservedTagKeyList(BaseModel): + """Represents a paginated list of reserved tag keys.""" + + model_config = ConfigDict(populate_by_name=True) + + items: list[ReservedTagKey] = Field( + default_factory=list, description="List of reserved tag keys" + ) + current_page: int | None = Field(None, description="Current page number") + total_pages: int | None = Field(None, description="Total number of pages") + prev_page: str | None = Field(None, description="URL of the previous page") + next_page: str | None = Field(None, description="URL of the next page") + total_count: int | None = Field(None, description="Total number of items") diff --git a/src/tfe/resources/reserved_tag_key.py b/src/tfe/resources/reserved_tag_key.py new file mode 100644 index 0000000..aeff161 --- /dev/null +++ b/src/tfe/resources/reserved_tag_key.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from typing import Any + +from ..errors import ( + InvalidOrgError, + ValidationError, +) +from ..models.reserved_tag_key import ( + ReservedTagKey as ReservedTagKeyModel, +) +from ..models.reserved_tag_key import ( + ReservedTagKeyCreateOptions, + ReservedTagKeyList, + ReservedTagKeyListOptions, + ReservedTagKeyUpdateOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class ReservedTagKey(_Service): + """Reserved Tag Key API for Terraform Enterprise.""" + + def list( + self, organization: str, options: ReservedTagKeyListOptions | None = None + ) -> ReservedTagKeyList: + """List reserved tag 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}/reserved-tag-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_reserved_tag_key(d)) + + return ReservedTagKeyList( + 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"), + ) + + def create( + self, organization: str, options: ReservedTagKeyCreateOptions + ) -> ReservedTagKeyModel: + """Create a new reserved tag key for the given organization.""" + if not valid_string_id(organization): + raise InvalidOrgError() + + attrs = options.model_dump(by_alias=True, exclude_none=True) + body: dict[str, Any] = { + "data": { + "attributes": attrs, + "type": "reserved-tag-keys", + } + } + + r = self.t.request( + "POST", + f"/api/v2/organizations/{organization}/reserved-tag-keys", + json_body=body, + ) + + jd = r.json() + data = jd.get("data", {}) + + return self._parse_reserved_tag_key(data) + + def read(self, reserved_tag_key_id: str) -> ReservedTagKeyModel: + """Read a reserved tag key by its ID.""" + if not valid_string_id(reserved_tag_key_id): + raise ValidationError("Invalid reserved tag key ID") + + # Note: Based on the API docs, there's no explicit GET endpoint for individual reserved tag keys + # This method would need to be implemented if such an endpoint exists + raise NotImplementedError( + "Individual reserved tag key read is not supported by the API" + ) + + def update( + self, reserved_tag_key_id: str, options: ReservedTagKeyUpdateOptions + ) -> ReservedTagKeyModel: + """Update a reserved tag key.""" + if not valid_string_id(reserved_tag_key_id): + raise ValidationError("Invalid reserved tag key ID") + + attrs = options.model_dump(by_alias=True, exclude_none=True) + body: dict[str, Any] = { + "data": { + "attributes": attrs, + "type": "reserved-tag-keys", + } + } + + r = self.t.request( + "PATCH", + f"/api/v2/reserved-tag-keys/{reserved_tag_key_id}", + json_body=body, + ) + + jd = r.json() + data = jd.get("data", {}) + + return self._parse_reserved_tag_key(data) + + def delete(self, reserved_tag_key_id: str) -> None: + """Delete a reserved tag key.""" + if not valid_string_id(reserved_tag_key_id): + raise ValidationError("Invalid reserved tag key ID") + + self.t.request("DELETE", f"/api/v2/reserved-tag-keys/{reserved_tag_key_id}") + # DELETE returns 204 No Content on success + + def _parse_reserved_tag_key(self, data: dict[str, Any]) -> ReservedTagKeyModel: + """Parse reserved tag key data from API response.""" + attrs = data.get("attributes", {}) + attrs["id"] = data.get("id") + return ReservedTagKeyModel.model_validate(attrs) diff --git a/tests/units/test_reserved_tag_key.py b/tests/units/test_reserved_tag_key.py new file mode 100644 index 0000000..318d11a --- /dev/null +++ b/tests/units/test_reserved_tag_key.py @@ -0,0 +1,120 @@ +"""Test the Reserved Tag Keys functionality.""" + +from unittest.mock import Mock + +import pytest + +from src.tfe._http import HTTPTransport +from src.tfe.errors import ( + InvalidOrgError, + ValidationError, +) +from src.tfe.models.reserved_tag_key import ( + ReservedTagKeyCreateOptions, + ReservedTagKeyListOptions, + ReservedTagKeyUpdateOptions, +) +from src.tfe.resources.reserved_tag_key import ReservedTagKey + + +class TestReservedTagKeyParsing: + """Test the reserved tag key parsing functionality.""" + + @pytest.fixture + def reserved_tag_key_service(self): + """Create a ReservedTagKey service for testing parsing.""" + mock_transport = Mock(spec=HTTPTransport) + return ReservedTagKey(mock_transport) + + def test_parse_reserved_tag_key_minimal(self, reserved_tag_key_service): + """Test _parse_reserved_tag_key with minimal data.""" + data = { + "id": "rtk-123", + "type": "reserved-tag-keys", + "attributes": { + "key": "environment", + "disable-overrides": False, + }, + } + reserved_tag_key = reserved_tag_key_service._parse_reserved_tag_key(data) + assert reserved_tag_key.id == "rtk-123" + assert reserved_tag_key.key == "environment" + assert reserved_tag_key.disable_overrides is False + + def test_parse_reserved_tag_key_with_dates(self, reserved_tag_key_service): + """Test _parse_reserved_tag_key with created_at and updated_at.""" + data = { + "id": "rtk-456", + "type": "reserved-tag-keys", + "attributes": { + "key": "cost-center", + "disable-overrides": True, + "created-at": "2024-08-13T23:06:42.523Z", + "updated-at": "2024-08-13T23:06:42.523Z", + }, + } + reserved_tag_key = reserved_tag_key_service._parse_reserved_tag_key(data) + assert reserved_tag_key.id == "rtk-456" + assert reserved_tag_key.key == "cost-center" + assert reserved_tag_key.disable_overrides is True + assert reserved_tag_key.created_at is not None + assert reserved_tag_key.updated_at is not None + + +class TestReservedTagKey: + """Test the Reserved Tag Key service.""" + + @pytest.fixture + def reserved_tag_key_service(self): + """Create a ReservedTagKey service for testing.""" + mock_transport = Mock(spec=HTTPTransport) + return ReservedTagKey(mock_transport) + + def test_list_reserved_tag_keys_invalid_org(self, reserved_tag_key_service): + """Test listing reserved tag keys with invalid organization.""" + with pytest.raises(InvalidOrgError): + reserved_tag_key_service.list("") + + def test_create_reserved_tag_key_invalid_org(self, reserved_tag_key_service): + """Test creating reserved tag key with invalid organization.""" + options = ReservedTagKeyCreateOptions( + key="environment", disable_overrides=False + ) + with pytest.raises(InvalidOrgError): + reserved_tag_key_service.create("", options) + + def test_read_reserved_tag_key_not_implemented(self, reserved_tag_key_service): + """Test reading reserved tag key raises NotImplementedError.""" + with pytest.raises(NotImplementedError): + reserved_tag_key_service.read("rtk-123") + + def test_update_reserved_tag_key_invalid_id(self, reserved_tag_key_service): + """Test updating reserved tag key with invalid ID.""" + options = ReservedTagKeyUpdateOptions(key="updated-key") + with pytest.raises(ValidationError): + reserved_tag_key_service.update("", options) + + def test_delete_reserved_tag_key_invalid_id(self, reserved_tag_key_service): + """Test deleting reserved tag key with invalid ID.""" + with pytest.raises(ValidationError): + reserved_tag_key_service.delete("") + + def test_reserved_tag_key_create_options_model(self): + """Test ReservedTagKeyCreateOptions model validation.""" + options = ReservedTagKeyCreateOptions(key="environment", disable_overrides=True) + assert options.key == "environment" + assert options.disable_overrides is True + + def test_reserved_tag_key_update_options_model(self): + """Test ReservedTagKeyUpdateOptions model validation.""" + options = ReservedTagKeyUpdateOptions( + key="updated-environment", disable_overrides=False + ) + assert options.key == "updated-environment" + assert options.disable_overrides is False + + def test_reserved_tag_key_list_options_model(self): + """Test ReservedTagKeyListOptions model validation.""" + options = ReservedTagKeyListOptions(page_number=2, page_size=50) + assert options.page_number == 2 + assert options.page_size == 50