From ea8f123d17daf49ef8c65d6b4d6898e6b516e11e Mon Sep 17 00:00:00 2001 From: aayushsingh2502 Date: Fri, 12 Sep 2025 17:58:25 +0530 Subject: [PATCH 1/5] workspace variables functions updates --- examples/variables.py | 296 ++++++++++++++++++++++++++++++++++ src/tfe/client.py | 2 + src/tfe/errors.py | 4 + src/tfe/resources/variable.py | 139 ++++++++++++++++ src/tfe/types.py | 42 +++++ 5 files changed, 483 insertions(+) create mode 100644 examples/variables.py create mode 100644 src/tfe/resources/variable.py diff --git a/examples/variables.py b/examples/variables.py new file mode 100644 index 0000000..9a4ffac --- /dev/null +++ b/examples/variables.py @@ -0,0 +1,296 @@ +#!/usr/bin/env python3 +""" +Comprehensive example testing all variable functions in TFE workspace. +Tests: list, list_all, create, read, update, and delete operations. +""" + +import sys +import os +import time + +# Add the src directory to the path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) + +from tfe import TFEClient, TFEConfig +from tfe.types import VariableCreateOptions, VariableUpdateOptions, CategoryType + + +def main(): + """Test all variable operations in a workspace.""" + + # 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 + + print(f"Testing all variable operations in workspace: {workspace_id}") + print("=" * 60) + + # Track created variables for cleanup + created_variables = [] + + try: + # 1. Test CREATE function - COMMENTED OUT (already have variables from previous run) + print("\n1. Testing CREATE operation:") + print("-" * 30) + + # Create a Terraform variable + terraform_var = VariableCreateOptions( + key="test_terraform_var", + value="production", + description="Test Terraform variable", + category=CategoryType.TERRAFORM, + hcl=False, + sensitive=False + ) + + try: + variable = client.variables.create(workspace_id, terraform_var) + created_variables.append(variable.id) + print(f"✓ Created Terraform variable: {variable.key} = {variable.value}") + print(f" ID: {variable.id}, Category: {variable.category}") + except Exception as e: + print(f"✗ Error creating Terraform variable: {e}") + + # Create an environment variable + env_var = VariableCreateOptions( + key="TEST_LOG_LEVEL", + value="DEBUG", + description="Test environment variable", + category=CategoryType.ENV, + hcl=False, + sensitive=False + ) + + try: + variable = client.variables.create(workspace_id, env_var) + created_variables.append(variable.id) + print(f"✓ Created environment variable: {variable.key} = {variable.value}") + print(f" ID: {variable.id}, Category: {variable.category}") + except Exception as e: + print(f"✗ Error creating environment variable: {e}") + + # Create a sensitive variable + secret_var = VariableCreateOptions( + key="TEST_API_KEY", + value="super-secret-key-12345", + description="Test sensitive variable", + category=CategoryType.ENV, + hcl=False, + sensitive=True + ) + + try: + variable = client.variables.create(workspace_id, secret_var) + created_variables.append(variable.id) + print(f"✓ Created sensitive variable: {variable.key} = ***HIDDEN***") + print(f" ID: {variable.id}, Category: {variable.category}") + except Exception as e: + print(f"✗ Error creating sensitive variable: {e}") + + # Small delay to ensure variables are created + time.sleep(1) + + # 2. Test LIST function (workspace-only variables) - COMMENTED OUT + print("\n2. Testing LIST operation (workspace variables only):") + print("-" * 50) + + try: + variables = list(client.variables.list(workspace_id)) + print(f"Found {len(variables)} workspace variables:") + for var in variables: + value_display = "***SENSITIVE***" if var.sensitive else var.value + print(f" • {var.key} = {value_display} ({var.category}) [ID: {var.id}]") + except Exception as e: + print(f"✗ Error listing variables: {e}") + + # 3. Test LIST_ALL function (includes inherited variables from variable sets) + print("\n3. Testing LIST_ALL operation (includes variable sets):") + print("-" * 55) + + try: + all_variables = list(client.variables.list_all(workspace_id)) + print(f"Found {len(all_variables)} total variables (including inherited):") + for var in all_variables: + value_display = "***SENSITIVE***" if var.sensitive else var.value + print(f" • {var.key} = {value_display} ({var.category}) [ID: {var.id}]") + except Exception as e: + print(f"✗ Error listing all variables: {e}") + + # Test READ function with specific variable ID - COMMENTED OUT + print("\n4. Testing READ operation with specific variable ID:") + print("-" * 50) + + # Replace this with actual variable ID to test reading + test_variable_id = "var-example123456789" + print(f"Testing READ with variable ID: {test_variable_id}") + + try: + variable = client.variables.read(workspace_id, test_variable_id) + # For testing, show actual values even for sensitive variables + if variable.sensitive: + print(f"✓ Read variable: {variable.key} = {variable.value} (SENSITIVE)") + else: + print(f"✓ Read variable: {variable.key} = {variable.value}") + print(f" ID: {variable.id}") + print(f" Description: {variable.description}") + print(f" Category: {variable.category}") + print(f" HCL: {variable.hcl}") + print(f" Sensitive: {variable.sensitive}") + if hasattr(variable, 'version_id'): + print(f" Version ID: {variable.version_id}") + except Exception as e: + print(f"✗ Error reading variable {test_variable_id}: {e}") + + # Test UPDATE function with specific variable ID - COMMENTED OUT + print("\n5. Testing UPDATE operation with specific variable ID:") + print("-" * 55) + + # Replace this with actual variable ID to test updating + test_variable_id = "var-example123456789" + print(f"Testing UPDATE with variable ID: {test_variable_id}") + print(f"Setting value to: 'npe'") + + try: + # First read the current variable to get its details + current_var = client.variables.read(workspace_id, test_variable_id) + print(f"Current value: {current_var.value}") + print(f"Current key: {current_var.key}") + + # Update the variable value to "npe" + update_options = VariableUpdateOptions( + key=current_var.key, + value="npe", + description=current_var.description, + hcl=current_var.hcl, + sensitive=current_var.sensitive + ) + + updated_variable = client.variables.update(workspace_id, test_variable_id, update_options) + print(f"✓ Updated variable: {updated_variable.key} = {updated_variable.value}") + print(f" Description: {updated_variable.description}") + print(f" Category: {updated_variable.category}") + print(f" HCL: {updated_variable.hcl}") + print(f" Sensitive: {updated_variable.sensitive}") + print(f" ID: {updated_variable.id}") + except Exception as e: + print(f"✗ Error updating variable {test_variable_id}: {e}") + + # Test DELETE function with specific variable ID + print("\n6. Testing DELETE operation with specific variable ID:") + print("-" * 55) + + # Replace this with actual variable ID to test deletion + test_variable_id = "var-example123456789" + print(f"Testing DELETE with variable ID: {test_variable_id}") + + try: + # First read the variable to confirm it exists before deletion + variable = client.variables.read(workspace_id, test_variable_id) + print(f"Variable to delete: {variable.key} = {variable.value}") + print(f" ID: {variable.id}") + + # Delete the variable + client.variables.delete(workspace_id, test_variable_id) + print(f"✓ Successfully deleted variable with ID: {test_variable_id}") + + # Try to read it again to verify deletion + print("Verifying deletion...") + try: + client.variables.read(workspace_id, test_variable_id) + print("✗ Warning: Variable still exists after deletion!") + except Exception as read_error: + if "not found" in str(read_error).lower() or "404" in str(read_error): + print("✓ Confirmed: Variable no longer exists") + else: + print(f"✗ Unexpected error verifying deletion: {read_error}") + + except Exception as e: + print(f"✗ Error deleting variable {test_variable_id}: {e}") + + + # 4. Test READ function + print("\n4. Testing READ operation:") + print("-" * 25) + + if created_variables: + test_var_id = created_variables[0] # Use the first created variable + try: + variable = client.variables.read(workspace_id, test_var_id) + value_display = "***SENSITIVE***" if variable.sensitive else variable.value + print(f"✓ Read variable: {variable.key} = {value_display}") + print(f" ID: {variable.id}") + print(f" Description: {variable.description}") + print(f" Category: {variable.category}") + print(f" HCL: {variable.hcl}") + print(f" Sensitive: {variable.sensitive}") + except Exception as e: + print(f"✗ Error reading variable {test_var_id}: {e}") + else: + print("No variables available to read") + + # 5. Test UPDATE function + print("\n5. Testing UPDATE operation:") + print("-" * 27) + + if created_variables: + test_var_id = created_variables[0] # Use the first created variable + try: + # First read the current variable to get its details + current_var = client.variables.read(workspace_id, test_var_id) + + # Update the variable + update_options = VariableUpdateOptions( + key=current_var.key, + value="updated_value_123", + description="Updated test variable description", + hcl=False, + sensitive=False + ) + + updated_variable = client.variables.update(workspace_id, test_var_id, update_options) + print(f"✓ Updated variable: {updated_variable.key} = {updated_variable.value}") + print(f" New description: {updated_variable.description}") + print(f" ID: {updated_variable.id}") + except Exception as e: + print(f"✗ Error updating variable {test_var_id}: {e}") + else: + print("No variables available to update") + + # 6. Test DELETE function + print("\n6. Testing DELETE operation:") + print("-" * 27) + + # Delete all created variables + for var_id in created_variables: + try: + client.variables.delete(workspace_id, var_id) + print(f"✓ Deleted variable with ID: {var_id}") + except Exception as e: + print(f"✗ Error deleting variable {var_id}: {e}") + + # Verify deletion by listing variables again + print("\nVerifying deletion - listing variables after cleanup:") + try: + remaining_variables = list(client.variables.list(workspace_id)) + # Filter out the variables we just deleted + remaining_test_vars = [v for v in remaining_variables if v.key.startswith("test_") or v.key.startswith("TEST_")] + if remaining_test_vars: + print(f"Warning: {len(remaining_test_vars)} test variables still exist:") + for var in remaining_test_vars: + print(f" • {var.key} [ID: {var.id}]") + else: + print("✓ All test variables successfully deleted") + except Exception as e: + print(f"✗ Error verifying deletion: {e}") + + except Exception as e: + print(f"✗ Unexpected error during testing: {e}") + + print("\n" + "=" * 60) + print("Variable testing complete!") + + +if __name__ == "__main__": + main() diff --git a/src/tfe/client.py b/src/tfe/client.py index e4e2754..8def377 100644 --- a/src/tfe/client.py +++ b/src/tfe/client.py @@ -4,6 +4,7 @@ from .config import TFEConfig from .resources.organizations import Organizations from .resources.projects import Projects +from .resources.variable import Variables from .resources.workspaces import Workspaces @@ -26,6 +27,7 @@ def __init__(self, config: TFEConfig | None = None): ) self.organizations = Organizations(self._transport) self.projects = Projects(self._transport) + self.variables = Variables(self._transport) self.workspaces = Workspaces(self._transport) def close(self) -> None: diff --git a/src/tfe/errors.py b/src/tfe/errors.py index e524f32..6c46acb 100644 --- a/src/tfe/errors.py +++ b/src/tfe/errors.py @@ -57,3 +57,7 @@ class RequiredFieldMissing(TFEError): ... ERR_REQUIRED_NAME = "name is required" ERR_INVALID_ORG = "invalid organization name" ERR_REQUIRED_EMAIL = "email is required" +ERR_INVALID_WORKSPACE_ID = "invalid workspace ID" +ERR_INVALID_VARIABLE_ID = "invalid variable ID" +ERR_REQUIRED_KEY = "key is required" +ERR_REQUIRED_CATEGORY = "category is required" diff --git a/src/tfe/resources/variable.py b/src/tfe/resources/variable.py new file mode 100644 index 0000000..f0fcafb --- /dev/null +++ b/src/tfe/resources/variable.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..errors import ( + ERR_INVALID_VARIABLE_ID, + ERR_INVALID_WORKSPACE_ID, + ERR_REQUIRED_CATEGORY, + ERR_REQUIRED_KEY, +) +from ..types import ( + Variable, + VariableCreateOptions, + VariableListOptions, + VariableUpdateOptions, +) +from ..utils import valid_string, valid_string_id +from ._base import _Service + + +class Variables(_Service): + def list(self, workspace_id: str, options: VariableListOptions | None = None) -> Iterator[Variable]: + """List all the variables associated with the given workspace (doesn't include variables inherited from varsets).""" + if not valid_string_id(workspace_id): + raise ValueError(ERR_INVALID_WORKSPACE_ID) + + path = f"/api/v2/workspaces/{workspace_id}/vars" + params = {} + if options: + # Add any options if needed in the future + pass + + for item in self._list(path, params=params): + attr = item.get("attributes", {}) or {} + var_id = item.get("id", "") + variable_data = dict(attr) + variable_data["id"] = var_id + yield Variable(**variable_data) + + def list_all(self, workspace_id: str, options: VariableListOptions | None = None) -> Iterator[Variable]: + """ListAll the variables associated with the given workspace including variables inherited from varsets.""" + if not valid_string_id(workspace_id): + raise ValueError(ERR_INVALID_WORKSPACE_ID) + + path = f"/api/v2/workspaces/{workspace_id}/all-vars" + params = {} + if options: + # Add any options if needed in the future + pass + + for item in self._list(path, params=params): + attr = item.get("attributes", {}) or {} + var_id = item.get("id", "") + variable_data = dict(attr) + variable_data["id"] = var_id + yield Variable(**variable_data) + + def create(self, workspace_id: str, options: VariableCreateOptions) -> Variable: + """Create is used to create a new variable.""" + if not valid_string_id(workspace_id): + raise ValueError(ERR_INVALID_WORKSPACE_ID) + + # Validate required fields + if not valid_string(options.key): + raise ValueError(ERR_REQUIRED_KEY) + if options.category is None: + raise ValueError(ERR_REQUIRED_CATEGORY) + + body = { + "data": { + "type": "vars", + "attributes": options.model_dump(exclude_none=True), + } + } + + response = self.t.request("POST", f"/api/v2/workspaces/{workspace_id}/vars", json_body=body) + data = response.json()["data"] + + # Parse the response and create Variable object + attr = data.get("attributes", {}) or {} + variable_id = data.get("id", "") + variable_data = dict(attr) + variable_data["id"] = variable_id + + return Variable(**variable_data) + + def read(self, workspace_id: str, variable_id: str) -> Variable: + """Read a variable by its ID.""" + if not valid_string_id(workspace_id): + raise ValueError(ERR_INVALID_WORKSPACE_ID) + if not valid_string_id(variable_id): + raise ValueError(ERR_INVALID_VARIABLE_ID) + + response = self.t.request("GET", f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}") + data = response.json()["data"] + + # Parse the response and create Variable object + attr = data.get("attributes", {}) or {} + var_id = data.get("id", "") + variable_data = dict(attr) + variable_data["id"] = var_id + + return Variable(**variable_data) + + def update(self, workspace_id: str, variable_id: str, options: VariableUpdateOptions) -> Variable: + """Update values of an existing variable.""" + if not valid_string_id(workspace_id): + raise ValueError(ERR_INVALID_WORKSPACE_ID) + if not valid_string_id(variable_id): + raise ValueError(ERR_INVALID_VARIABLE_ID) + + body = { + "data": { + "type": "vars", + "attributes": options.model_dump(exclude_none=True), + } + } + + response = self.t.request("PATCH", f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}", json_body=body) + data = response.json()["data"] + + # Parse the response and create Variable object + attr = data.get("attributes", {}) or {} + var_id = data.get("id", "") + variable_data = dict(attr) + variable_data["id"] = var_id + + return Variable(**variable_data) + + def delete(self, workspace_id: str, variable_id: str) -> None: + """Delete a variable by its ID.""" + if not valid_string_id(workspace_id): + raise ValueError(ERR_INVALID_WORKSPACE_ID) + if not valid_string_id(variable_id): + raise ValueError(ERR_INVALID_VARIABLE_ID) + + self.t.request("DELETE", f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}") + return None diff --git a/src/tfe/types.py b/src/tfe/types.py index f5581e2..7f79524 100644 --- a/src/tfe/types.py +++ b/src/tfe/types.py @@ -222,3 +222,45 @@ class DataRetentionPolicyDeleteOlderSetOptions(BaseModel): class DataRetentionPolicyDontDeleteSetOptions(BaseModel): pass # No additional fields needed + + +# Variables related models +class CategoryType(str, Enum): + ENV = "env" + POLICY_SET = "policy-set" + TERRAFORM = "terraform" + + +class Variable(BaseModel): + id: str | None = None + key: str | None = None + value: str | None = None + description: str | None = None + category: CategoryType | None = None + hcl: bool | None = None + sensitive: bool | None = None + version_id: str | None = None + workspace: dict | None = None + + +class VariableListOptions(BaseModel): + # Base pagination options would be handled by the service layer + pass + + +class VariableCreateOptions(BaseModel): + key: str | None = None + value: str | None = None + description: str | None = None + category: CategoryType | None = None + hcl: bool | None = None + sensitive: bool | None = None + + +class VariableUpdateOptions(BaseModel): + key: str | None = None + value: str | None = None + description: str | None = None + category: CategoryType | None = None + hcl: bool | None = None + sensitive: bool | None = None From d489135d418cb0d2384f39e386a66393be550100 Mon Sep 17 00:00:00 2001 From: aayushsingh2502 Date: Fri, 12 Sep 2025 18:07:37 +0530 Subject: [PATCH 2/5] code clean and fmt --- examples/variables.py | 141 +++++++++++++++++++--------------- src/tfe/resources/variable.py | 65 ++++++++++------ 2 files changed, 120 insertions(+), 86 deletions(-) diff --git a/examples/variables.py b/examples/variables.py index 9a4ffac..b3987d9 100644 --- a/examples/variables.py +++ b/examples/variables.py @@ -4,37 +4,37 @@ Tests: list, list_all, create, read, update, and delete operations. """ -import sys import os +import sys import time # Add the src directory to the path -sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) from tfe import TFEClient, TFEConfig -from tfe.types import VariableCreateOptions, VariableUpdateOptions, CategoryType +from tfe.types import CategoryType, VariableCreateOptions, VariableUpdateOptions def main(): """Test all variable operations in a workspace.""" - + # 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 - + print(f"Testing all variable operations in workspace: {workspace_id}") print("=" * 60) - + # Track created variables for cleanup created_variables = [] - + try: # 1. Test CREATE function - COMMENTED OUT (already have variables from previous run) print("\n1. Testing CREATE operation:") print("-" * 30) - + # Create a Terraform variable terraform_var = VariableCreateOptions( key="test_terraform_var", @@ -42,9 +42,9 @@ def main(): description="Test Terraform variable", category=CategoryType.TERRAFORM, hcl=False, - sensitive=False + sensitive=False, ) - + try: variable = client.variables.create(workspace_id, terraform_var) created_variables.append(variable.id) @@ -52,7 +52,7 @@ def main(): print(f" ID: {variable.id}, Category: {variable.category}") except Exception as e: print(f"✗ Error creating Terraform variable: {e}") - + # Create an environment variable env_var = VariableCreateOptions( key="TEST_LOG_LEVEL", @@ -60,9 +60,9 @@ def main(): description="Test environment variable", category=CategoryType.ENV, hcl=False, - sensitive=False + sensitive=False, ) - + try: variable = client.variables.create(workspace_id, env_var) created_variables.append(variable.id) @@ -70,7 +70,7 @@ def main(): print(f" ID: {variable.id}, Category: {variable.category}") except Exception as e: print(f"✗ Error creating environment variable: {e}") - + # Create a sensitive variable secret_var = VariableCreateOptions( key="TEST_API_KEY", @@ -78,9 +78,9 @@ def main(): description="Test sensitive variable", category=CategoryType.ENV, hcl=False, - sensitive=True + sensitive=True, ) - + try: variable = client.variables.create(workspace_id, secret_var) created_variables.append(variable.id) @@ -88,44 +88,48 @@ def main(): print(f" ID: {variable.id}, Category: {variable.category}") except Exception as e: print(f"✗ Error creating sensitive variable: {e}") - + # Small delay to ensure variables are created time.sleep(1) - + # 2. Test LIST function (workspace-only variables) - COMMENTED OUT print("\n2. Testing LIST operation (workspace variables only):") print("-" * 50) - + try: variables = list(client.variables.list(workspace_id)) print(f"Found {len(variables)} workspace variables:") for var in variables: value_display = "***SENSITIVE***" if var.sensitive else var.value - print(f" • {var.key} = {value_display} ({var.category}) [ID: {var.id}]") + print( + f" • {var.key} = {value_display} ({var.category}) [ID: {var.id}]" + ) except Exception as e: print(f"✗ Error listing variables: {e}") - + # 3. Test LIST_ALL function (includes inherited variables from variable sets) print("\n3. Testing LIST_ALL operation (includes variable sets):") print("-" * 55) - + try: all_variables = list(client.variables.list_all(workspace_id)) print(f"Found {len(all_variables)} total variables (including inherited):") for var in all_variables: value_display = "***SENSITIVE***" if var.sensitive else var.value - print(f" • {var.key} = {value_display} ({var.category}) [ID: {var.id}]") + print( + f" • {var.key} = {value_display} ({var.category}) [ID: {var.id}]" + ) except Exception as e: print(f"✗ Error listing all variables: {e}") - + # Test READ function with specific variable ID - COMMENTED OUT print("\n4. Testing READ operation with specific variable ID:") print("-" * 50) - + # Replace this with actual variable ID to test reading test_variable_id = "var-example123456789" print(f"Testing READ with variable ID: {test_variable_id}") - + try: variable = client.variables.read(workspace_id, test_variable_id) # For testing, show actual values even for sensitive variables @@ -138,37 +142,41 @@ def main(): print(f" Category: {variable.category}") print(f" HCL: {variable.hcl}") print(f" Sensitive: {variable.sensitive}") - if hasattr(variable, 'version_id'): + if hasattr(variable, "version_id"): print(f" Version ID: {variable.version_id}") except Exception as e: print(f"✗ Error reading variable {test_variable_id}: {e}") - + # Test UPDATE function with specific variable ID - COMMENTED OUT print("\n5. Testing UPDATE operation with specific variable ID:") print("-" * 55) - + # Replace this with actual variable ID to test updating test_variable_id = "var-example123456789" print(f"Testing UPDATE with variable ID: {test_variable_id}") - print(f"Setting value to: 'npe'") - + print("Setting value to: 'npe'") + try: # First read the current variable to get its details current_var = client.variables.read(workspace_id, test_variable_id) print(f"Current value: {current_var.value}") print(f"Current key: {current_var.key}") - + # Update the variable value to "npe" update_options = VariableUpdateOptions( key=current_var.key, value="npe", description=current_var.description, hcl=current_var.hcl, - sensitive=current_var.sensitive + sensitive=current_var.sensitive, + ) + + updated_variable = client.variables.update( + workspace_id, test_variable_id, update_options + ) + print( + f"✓ Updated variable: {updated_variable.key} = {updated_variable.value}" ) - - updated_variable = client.variables.update(workspace_id, test_variable_id, update_options) - print(f"✓ Updated variable: {updated_variable.key} = {updated_variable.value}") print(f" Description: {updated_variable.description}") print(f" Category: {updated_variable.category}") print(f" HCL: {updated_variable.hcl}") @@ -176,25 +184,25 @@ def main(): print(f" ID: {updated_variable.id}") except Exception as e: print(f"✗ Error updating variable {test_variable_id}: {e}") - + # Test DELETE function with specific variable ID print("\n6. Testing DELETE operation with specific variable ID:") print("-" * 55) - + # Replace this with actual variable ID to test deletion test_variable_id = "var-example123456789" print(f"Testing DELETE with variable ID: {test_variable_id}") - + try: # First read the variable to confirm it exists before deletion variable = client.variables.read(workspace_id, test_variable_id) print(f"Variable to delete: {variable.key} = {variable.value}") print(f" ID: {variable.id}") - + # Delete the variable client.variables.delete(workspace_id, test_variable_id) print(f"✓ Successfully deleted variable with ID: {test_variable_id}") - + # Try to read it again to verify deletion print("Verifying deletion...") try: @@ -205,20 +213,21 @@ def main(): print("✓ Confirmed: Variable no longer exists") else: print(f"✗ Unexpected error verifying deletion: {read_error}") - + except Exception as e: print(f"✗ Error deleting variable {test_variable_id}: {e}") - - + # 4. Test READ function print("\n4. Testing READ operation:") print("-" * 25) - + if created_variables: test_var_id = created_variables[0] # Use the first created variable try: variable = client.variables.read(workspace_id, test_var_id) - value_display = "***SENSITIVE***" if variable.sensitive else variable.value + value_display = ( + "***SENSITIVE***" if variable.sensitive else variable.value + ) print(f"✓ Read variable: {variable.key} = {value_display}") print(f" ID: {variable.id}") print(f" Description: {variable.description}") @@ -229,39 +238,43 @@ def main(): print(f"✗ Error reading variable {test_var_id}: {e}") else: print("No variables available to read") - + # 5. Test UPDATE function print("\n5. Testing UPDATE operation:") print("-" * 27) - + if created_variables: test_var_id = created_variables[0] # Use the first created variable try: # First read the current variable to get its details current_var = client.variables.read(workspace_id, test_var_id) - + # Update the variable update_options = VariableUpdateOptions( key=current_var.key, value="updated_value_123", description="Updated test variable description", hcl=False, - sensitive=False + sensitive=False, + ) + + updated_variable = client.variables.update( + workspace_id, test_var_id, update_options + ) + print( + f"✓ Updated variable: {updated_variable.key} = {updated_variable.value}" ) - - updated_variable = client.variables.update(workspace_id, test_var_id, update_options) - print(f"✓ Updated variable: {updated_variable.key} = {updated_variable.value}") print(f" New description: {updated_variable.description}") print(f" ID: {updated_variable.id}") except Exception as e: print(f"✗ Error updating variable {test_var_id}: {e}") else: print("No variables available to update") - + # 6. Test DELETE function print("\n6. Testing DELETE operation:") print("-" * 27) - + # Delete all created variables for var_id in created_variables: try: @@ -269,25 +282,31 @@ def main(): print(f"✓ Deleted variable with ID: {var_id}") except Exception as e: print(f"✗ Error deleting variable {var_id}: {e}") - + # Verify deletion by listing variables again print("\nVerifying deletion - listing variables after cleanup:") try: remaining_variables = list(client.variables.list(workspace_id)) # Filter out the variables we just deleted - remaining_test_vars = [v for v in remaining_variables if v.key.startswith("test_") or v.key.startswith("TEST_")] + remaining_test_vars = [ + v + for v in remaining_variables + if v.key.startswith("test_") or v.key.startswith("TEST_") + ] if remaining_test_vars: - print(f"Warning: {len(remaining_test_vars)} test variables still exist:") + print( + f"Warning: {len(remaining_test_vars)} test variables still exist:" + ) for var in remaining_test_vars: print(f" • {var.key} [ID: {var.id}]") else: print("✓ All test variables successfully deleted") except Exception as e: print(f"✗ Error verifying deletion: {e}") - + except Exception as e: print(f"✗ Unexpected error during testing: {e}") - + print("\n" + "=" * 60) print("Variable testing complete!") diff --git a/src/tfe/resources/variable.py b/src/tfe/resources/variable.py index f0fcafb..d4ede38 100644 --- a/src/tfe/resources/variable.py +++ b/src/tfe/resources/variable.py @@ -1,7 +1,6 @@ from __future__ import annotations from collections.abc import Iterator -from typing import Any from ..errors import ( ERR_INVALID_VARIABLE_ID, @@ -20,17 +19,19 @@ class Variables(_Service): - def list(self, workspace_id: str, options: VariableListOptions | None = None) -> Iterator[Variable]: + def list( + self, workspace_id: str, options: VariableListOptions | None = None + ) -> Iterator[Variable]: """List all the variables associated with the given workspace (doesn't include variables inherited from varsets).""" if not valid_string_id(workspace_id): raise ValueError(ERR_INVALID_WORKSPACE_ID) - + path = f"/api/v2/workspaces/{workspace_id}/vars" params = {} if options: # Add any options if needed in the future pass - + for item in self._list(path, params=params): attr = item.get("attributes", {}) or {} var_id = item.get("id", "") @@ -38,17 +39,19 @@ def list(self, workspace_id: str, options: VariableListOptions | None = None) -> variable_data["id"] = var_id yield Variable(**variable_data) - def list_all(self, workspace_id: str, options: VariableListOptions | None = None) -> Iterator[Variable]: + def list_all( + self, workspace_id: str, options: VariableListOptions | None = None + ) -> Iterator[Variable]: """ListAll the variables associated with the given workspace including variables inherited from varsets.""" if not valid_string_id(workspace_id): raise ValueError(ERR_INVALID_WORKSPACE_ID) - + path = f"/api/v2/workspaces/{workspace_id}/all-vars" params = {} if options: # Add any options if needed in the future pass - + for item in self._list(path, params=params): attr = item.get("attributes", {}) or {} var_id = item.get("id", "") @@ -60,29 +63,31 @@ def create(self, workspace_id: str, options: VariableCreateOptions) -> Variable: """Create is used to create a new variable.""" if not valid_string_id(workspace_id): raise ValueError(ERR_INVALID_WORKSPACE_ID) - + # Validate required fields if not valid_string(options.key): raise ValueError(ERR_REQUIRED_KEY) if options.category is None: raise ValueError(ERR_REQUIRED_CATEGORY) - + body = { "data": { "type": "vars", "attributes": options.model_dump(exclude_none=True), } } - - response = self.t.request("POST", f"/api/v2/workspaces/{workspace_id}/vars", json_body=body) + + response = self.t.request( + "POST", f"/api/v2/workspaces/{workspace_id}/vars", json_body=body + ) data = response.json()["data"] - + # Parse the response and create Variable object attr = data.get("attributes", {}) or {} variable_id = data.get("id", "") variable_data = dict(attr) variable_data["id"] = variable_id - + return Variable(**variable_data) def read(self, workspace_id: str, variable_id: str) -> Variable: @@ -91,41 +96,49 @@ def read(self, workspace_id: str, variable_id: str) -> Variable: raise ValueError(ERR_INVALID_WORKSPACE_ID) if not valid_string_id(variable_id): raise ValueError(ERR_INVALID_VARIABLE_ID) - - response = self.t.request("GET", f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}") + + response = self.t.request( + "GET", f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}" + ) data = response.json()["data"] - + # Parse the response and create Variable object attr = data.get("attributes", {}) or {} var_id = data.get("id", "") variable_data = dict(attr) variable_data["id"] = var_id - + return Variable(**variable_data) - def update(self, workspace_id: str, variable_id: str, options: VariableUpdateOptions) -> Variable: + def update( + self, workspace_id: str, variable_id: str, options: VariableUpdateOptions + ) -> Variable: """Update values of an existing variable.""" if not valid_string_id(workspace_id): raise ValueError(ERR_INVALID_WORKSPACE_ID) if not valid_string_id(variable_id): raise ValueError(ERR_INVALID_VARIABLE_ID) - + body = { "data": { "type": "vars", "attributes": options.model_dump(exclude_none=True), } } - - response = self.t.request("PATCH", f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}", json_body=body) + + response = self.t.request( + "PATCH", + f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}", + json_body=body, + ) data = response.json()["data"] - + # Parse the response and create Variable object attr = data.get("attributes", {}) or {} var_id = data.get("id", "") variable_data = dict(attr) variable_data["id"] = var_id - + return Variable(**variable_data) def delete(self, workspace_id: str, variable_id: str) -> None: @@ -134,6 +147,8 @@ def delete(self, workspace_id: str, variable_id: str) -> None: raise ValueError(ERR_INVALID_WORKSPACE_ID) if not valid_string_id(variable_id): raise ValueError(ERR_INVALID_VARIABLE_ID) - - self.t.request("DELETE", f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}") + + self.t.request( + "DELETE", f"/api/v2/workspaces/{workspace_id}/vars/{variable_id}" + ) return None From 54cb21b58966ab9d7c3f31c39c22be23ab1c5d1d Mon Sep 17 00:00:00 2001 From: aayushsingh2502 Date: Fri, 12 Sep 2025 18:16:39 +0530 Subject: [PATCH 3/5] lint issue fix --- src/tfe/resources/variable.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/tfe/resources/variable.py b/src/tfe/resources/variable.py index d4ede38..b6751f5 100644 --- a/src/tfe/resources/variable.py +++ b/src/tfe/resources/variable.py @@ -1,6 +1,7 @@ from __future__ import annotations from collections.abc import Iterator +from typing import Any from ..errors import ( ERR_INVALID_VARIABLE_ID, @@ -27,7 +28,7 @@ def list( raise ValueError(ERR_INVALID_WORKSPACE_ID) path = f"/api/v2/workspaces/{workspace_id}/vars" - params = {} + params: dict[str, Any] = {} if options: # Add any options if needed in the future pass @@ -47,7 +48,7 @@ def list_all( raise ValueError(ERR_INVALID_WORKSPACE_ID) path = f"/api/v2/workspaces/{workspace_id}/all-vars" - params = {} + params: dict[str, Any] = {} if options: # Add any options if needed in the future pass From cb0b9749e2fff3ea88837820e7e9a7126818bb82 Mon Sep 17 00:00:00 2001 From: Prabuddha Chakraborty Date: Wed, 17 Sep 2025 20:13:21 +0530 Subject: [PATCH 4/5] Fix conflicts from UI --- src/tfe/types.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/tfe/types.py b/src/tfe/types.py index 96ffe73..707a06c 100644 --- a/src/tfe/types.py +++ b/src/tfe/types.py @@ -326,7 +326,7 @@ class VariableUpdateOptions(BaseModel): category: CategoryType | None = None hcl: bool | None = None sensitive: bool | None = None -======= + class Tag(BaseModel): id: str | None = None name: str = "" From ebfbbe73a6bf1e3b555c98ea4bd3c6019b4d2f6f Mon Sep 17 00:00:00 2001 From: Prabuddha Chakraborty Date: Wed, 17 Sep 2025 20:16:18 +0530 Subject: [PATCH 5/5] Lint checks update --- src/tfe/types.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/tfe/types.py b/src/tfe/types.py index 707a06c..f7738d7 100644 --- a/src/tfe/types.py +++ b/src/tfe/types.py @@ -286,6 +286,7 @@ class DataRetentionPolicyDeleteOlderSetOptions(BaseModel): class DataRetentionPolicyDontDeleteSetOptions(BaseModel): pass # No additional fields needed + # Variables related models class CategoryType(str, Enum): ENV = "env" @@ -327,6 +328,7 @@ class VariableUpdateOptions(BaseModel): hcl: bool | None = None sensitive: bool | None = None + class Tag(BaseModel): id: str | None = None name: str = ""