From 113cbb22000d9139cf1238474c609e4f1a2c1502 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Mon, 24 Nov 2025 17:02:56 +0530 Subject: [PATCH 1/6] feat(policy set parameter): added list, create and read methods --- examples/policy_set_parameter.py | 122 +++++++++++++++++ src/pytfe/client.py | 2 + src/pytfe/errors.py | 29 ++++ src/pytfe/models/__init__.py | 13 ++ src/pytfe/models/policy_set_parameter.py | 56 ++++++++ src/pytfe/resources/policy_set_parameter.py | 138 ++++++++++++++++++++ 6 files changed, 360 insertions(+) create mode 100644 examples/policy_set_parameter.py create mode 100644 src/pytfe/models/policy_set_parameter.py create mode 100644 src/pytfe/resources/policy_set_parameter.py diff --git a/examples/policy_set_parameter.py b/examples/policy_set_parameter.py new file mode 100644 index 00000000..e202cf89 --- /dev/null +++ b/examples/policy_set_parameter.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +import argparse +import os + +from pytfe import TFEClient, TFEConfig +from pytfe.models import ( + PolicySetParameterCreateOptions, + PolicySetParameterListOptions, +) + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser( + description="Policy Set Parameters 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("--policy-set-id", required=True, help="Policy Set ID") + parser.add_argument("--page", type=int, default=1) + parser.add_argument("--page-size", type=int, default=10) + parser.add_argument("--create", action="store_true", help="Create a test parameter") + parser.add_argument("--read", action="store_true", help="Read a specific parameter") + parser.add_argument("--parameter-id", help="Parameter ID for read operation") + parser.add_argument( + "--key", default="test_param", help="Parameter key for creation" + ) + parser.add_argument( + "--value", default="test_value", help="Parameter value for creation" + ) + parser.add_argument( + "--sensitive", action="store_true", help="Mark parameter as sensitive" + ) + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + # 1) List all parameters for the policy set + _print_header(f"Listing parameters for policy set: {args.policy_set_id}") + + options = PolicySetParameterListOptions( + page_number=args.page, + page_size=args.page_size, + ) + + param_list = client.policy_set_parameters.list(args.policy_set_id, options) + + print(f"Total parameters: {param_list.total_count}") + print(f"Page {param_list.current_page} of {param_list.total_pages}") + print() + + if not param_list.items: + print("No parameters found.") + else: + for param in param_list.items: + # Sensitive parameters will have masked values + value_display = "***SENSITIVE***" if param.sensitive else param.value + print(f"- {param.id}") + print(f" Key: {param.key}") + print(f" Value: {value_display}") + print(f" Category: {param.category.value}") + print(f" Sensitive: {param.sensitive}") + print() + + # 2) Read a specific parameter (if --read flag is provided) + if args.read: + if not args.parameter_id: + print("Error: --parameter-id is required for read operation") + return + + _print_header(f"Reading parameter: {args.parameter_id}") + + param = client.policy_set_parameters.read(args.policy_set_id, args.parameter_id) + + print(f"Parameter ID: {param.id}") + print(f" Key: {param.key}") + value_display = "***SENSITIVE***" if param.sensitive else param.value + print(f" Value: {value_display}") + print(f" Category: {param.category.value}") + print(f" Sensitive: {param.sensitive}") + + # 3) Create a new parameter (if --create flag is provided) + if args.create: + _print_header(f"Creating new parameter with key: {args.key}") + + create_options = PolicySetParameterCreateOptions( + key=args.key, + value=args.value, + sensitive=args.sensitive, + ) + + new_param = client.policy_set_parameters.create( + args.policy_set_id, create_options + ) + + print(f"Created parameter: {new_param.id}") + print(f" Key: {new_param.key}") + value_display = "***SENSITIVE***" if new_param.sensitive else new_param.value + print(f" Value: {value_display}") + print(f" Category: {new_param.category.value}") + print(f" Sensitive: {new_param.sensitive}") + + # List again to show the new parameter + _print_header("Listing parameters after creation") + updated_list = client.policy_set_parameters.list(args.policy_set_id) + print(f"Total parameters: {updated_list.total_count}") + for param in updated_list.items: + value_display = "***SENSITIVE***" if param.sensitive else param.value + print(f"- {param.key}: {value_display} (sensitive={param.sensitive})") + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 38948863..4d0227d2 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -16,6 +16,7 @@ from .resources.policy_evaluation import PolicyEvaluations from .resources.policy_set import PolicySets from .resources.policy_set_outcome import PolicySets as PolicySetOutcomes +from .resources.policy_set_parameter import PolicySetParameters from .resources.policy_set_version import PolicySetVersions from .resources.projects import Projects from .resources.query_run import QueryRuns @@ -84,6 +85,7 @@ def __init__(self, config: TFEConfig | None = None): self.policy_evaluations = PolicyEvaluations(self._transport) self.policy_checks = PolicyChecks(self._transport) self.policy_sets = PolicySets(self._transport) + self.policy_set_parameters = PolicySetParameters(self._transport) self.policy_set_outcomes = PolicySetOutcomes(self._transport) self.policy_set_versions = PolicySetVersions(self._transport) diff --git a/src/pytfe/errors.py b/src/pytfe/errors.py index 3eac2be3..d3659048 100644 --- a/src/pytfe/errors.py +++ b/src/pytfe/errors.py @@ -460,3 +460,32 @@ class InvalidPolicyEvaluationIDError(InvalidValues): def __init__(self, message: str = "invalid value for policy evaluation ID"): super().__init__(message) + + +# Policy Set Parameter errors +class InvalidParamIDError(InvalidValues): + """Raised when an invalid policy set parameter ID is provided.""" + + def __init__(self, message: str = "invalid value for parameter ID"): + super().__init__(message) + + +class RequiredCategoryError(RequiredFieldMissing): + """Raised when a required category field is missing.""" + + def __init__(self, message: str = "category is required"): + super().__init__(message) + + +class InvalidCategoryError(InvalidValues): + """Raised when an invalid category field is provided.""" + + def __init__(self, message: str = "category must be policy-set"): + super().__init__(message) + + +class RequiredKeyError(RequiredFieldMissing): + """Raised when a required key field is missing.""" + + def __init__(self, message: str = "key is required"): + super().__init__(message) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index f3eb33bc..b19c4464 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -133,6 +133,13 @@ PolicySetRemoveWorkspacesOptions, PolicySetUpdateOptions, ) +from .policy_set_parameter import ( + PolicySetParameter, + PolicySetParameterCreateOptions, + PolicySetParameterList, + PolicySetParameterListOptions, + PolicySetParameterUpdateOptions, +) from .policy_types import ( EnforcementLevel, PolicyKind, @@ -586,6 +593,12 @@ "PolicySetRemoveWorkspaceExclusionsOptions", "PolicySetRemoveProjectsOptions", "PolicySetUpdateOptions", + # Policy Set Parameters + "PolicySetParameter", + "PolicySetParameterCreateOptions", + "PolicySetParameterList", + "PolicySetParameterListOptions", + "PolicySetParameterUpdateOptions", "PolicyKind", "EnforcementLevel", # Variable Sets diff --git a/src/pytfe/models/policy_set_parameter.py b/src/pytfe/models/policy_set_parameter.py new file mode 100644 index 00000000..8666a630 --- /dev/null +++ b/src/pytfe/models/policy_set_parameter.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + +from .policy_set import PolicySet +from .variable import CategoryType + + +class PolicySetParameter(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + key: str = Field(..., alias="key") + value: str | None = Field(None, alias="value") + category: CategoryType = Field(..., alias="category") + sensitive: bool = Field(..., alias="sensitive") + + # relations + policy_set: PolicySet = Field(..., alias="configurable") + + +class PolicySetParameterList(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + items: list[PolicySetParameter] = Field(default_factory=list) + current_page: int | None = None + total_pages: int | None = None + prev_page: int | None = None + next_page: int | None = None + total_count: int | None = None + + +class PolicySetParameterListOptions(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + page_number: int | None = Field(None, alias="page[number]") + page_size: int | None = Field(None, alias="page[size]") + + +class PolicySetParameterCreateOptions(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + key: str = Field(..., alias="key") + value: str | None = Field(None, alias="value") + + # Required: The Category of the parameter, should always be "policy-set" + category: CategoryType = Field(default=CategoryType.POLICY_SET, alias="category") + sensitive: bool | None = Field(None, alias="sensitive") + + +class PolicySetParameterUpdateOptions(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + key: str | None = Field(None, alias="key") + value: str | None = Field(None, alias="value") + sensitive: bool | None = Field(None, alias="sensitive") diff --git a/src/pytfe/resources/policy_set_parameter.py b/src/pytfe/resources/policy_set_parameter.py new file mode 100644 index 00000000..56fa9610 --- /dev/null +++ b/src/pytfe/resources/policy_set_parameter.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +from ..errors import ( + InvalidCategoryError, + InvalidParamIDError, + InvalidPolicySetIDError, + RequiredCategoryError, + RequiredKeyError, +) +from ..models.policy_set_parameter import ( + PolicySetParameter, + PolicySetParameterCreateOptions, + PolicySetParameterList, + PolicySetParameterListOptions, +) +from ..models.variable import CategoryType +from ..utils import valid_string, valid_string_id +from ._base import _Service + + +class PolicySetParameters(_Service): + """ + PolicySetParameters describes all the parameter related methods that the Terraform Enterprise API supports. + TFE API docs: https://developer.hashicorp.com/terraform/cloud-docs/api-docs/policy-set-params + """ + + def list( + self, policy_set_id: str, options: PolicySetParameterListOptions | None = None + ) -> PolicySetParameterList: + """List all the parameters associated with the given policy-set.""" + if not valid_string_id(policy_set_id): + raise InvalidPolicySetIDError() + params = options.model_dump(by_alias=True, exclude_none=True) if options else {} + r = self.t.request( + "GET", + path=f"api/v2/policy-sets/{policy_set_id}/parameters", + params=params, + ) + jd = r.json() + items = [] + meta = jd.get("meta", {}) + pagination = meta.get("pagination", {}) + for d in jd.get("data", []): + attrs = d.get("attributes", {}) + attrs["id"] = d.get("id") + attrs["policy_set"] = ( + d.get("relationships", {}).get("configurable", {}).get("data", {}) + ) + items.append(PolicySetParameter.model_validate(attrs)) + return PolicySetParameterList( + 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, policy_set_id: str, options: PolicySetParameterCreateOptions + ) -> PolicySetParameter: + """Create is used to create a new parameter.""" + if not valid_string_id(policy_set_id): + raise InvalidPolicySetIDError() + + if not valid_string(options.key): + raise RequiredKeyError() + + if options.category is None: + raise RequiredCategoryError() + if options.category != CategoryType.POLICY_SET: + raise InvalidCategoryError() + + attributes = options.model_dump(by_alias=True, exclude_none=True) + payload = { + "data": { + "type": "vars", + "attributes": attributes, + } + } + r = self.t.request( + "POST", + path=f"api/v2/policy-sets/{policy_set_id}/parameters", + json_body=payload, + ) + jd = r.json() + data = jd.get("data", {}) + attrs = data.get("attributes", {}) + attrs["id"] = data.get("id") + attrs["policy_set"] = ( + data.get("relationships", {}).get("configurable", {}).get("data", {}) + ) + return PolicySetParameter.model_validate(attrs) + + def read(self, policy_set_id: str, parameter_id: str) -> PolicySetParameter: + """Read a parameter by its ID.""" + if not valid_string_id(policy_set_id): + raise InvalidPolicySetIDError() + + if not valid_string_id(parameter_id): + raise InvalidParamIDError() + + r = self.t.request( + "GET", + path=f"api/v2/policy-sets/{policy_set_id}/parameters/{parameter_id}", + ) + jd = r.json() + data = jd.get("data", {}) + attrs = data.get("attributes", {}) + attrs["id"] = data.get("id") + attrs["policy_set"] = ( + data.get("relationships", {}).get("configurable", {}).get("data", {}) + ) + return PolicySetParameter.model_validate(attrs) + + +""" + def update( + self, + policy_set_id: str, + parameter_id: str, + options: PolicySetParameterUpdateOptions, + ) -> PolicySetParameter: + if not valid_string_id(policy_set_id): + raise InvalidPolicySetIDError() + + if not valid_string_id(parameter_id): + raise InvalidParamIDError() + return PolicySetParameter() + + def delete(self, policy_set_id: str, parameter_id: str) -> None: + if not valid_string_id(policy_set_id): + raise InvalidPolicySetIDError() + + if not valid_string_id(parameter_id): + raise InvalidParamIDError() + return None +""" From 97e32304f6b32b0e656712e7b09d85d2d3e8da76 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Mon, 24 Nov 2025 17:04:25 +0530 Subject: [PATCH 2/6] feat(policy set parameter): added update and delete methods --- examples/policy_set_parameter.py | 100 ++++++++++++++++++-- src/pytfe/resources/policy_set_parameter.py | 30 +++++- 2 files changed, 117 insertions(+), 13 deletions(-) diff --git a/examples/policy_set_parameter.py b/examples/policy_set_parameter.py index e202cf89..77302203 100644 --- a/examples/policy_set_parameter.py +++ b/examples/policy_set_parameter.py @@ -7,6 +7,7 @@ from pytfe.models import ( PolicySetParameterCreateOptions, PolicySetParameterListOptions, + PolicySetParameterUpdateOptions, ) @@ -29,13 +30,11 @@ def main(): parser.add_argument("--page-size", type=int, default=10) parser.add_argument("--create", action="store_true", help="Create a test parameter") parser.add_argument("--read", action="store_true", help="Read a specific parameter") - parser.add_argument("--parameter-id", help="Parameter ID for read operation") - parser.add_argument( - "--key", default="test_param", help="Parameter key for creation" - ) - parser.add_argument( - "--value", default="test_value", help="Parameter value for creation" - ) + parser.add_argument("--update", action="store_true", help="Update a parameter") + parser.add_argument("--delete", action="store_true", help="Delete a parameter") + parser.add_argument("--parameter-id", help="Parameter ID for read/update/delete operation") + parser.add_argument("--key", help="Parameter key for creation/update") + parser.add_argument("--value", help="Parameter value for creation/update") parser.add_argument( "--sensitive", action="store_true", help="Mark parameter as sensitive" ) @@ -88,13 +87,96 @@ def main(): print(f" Category: {param.category.value}") print(f" Sensitive: {param.sensitive}") - # 3) Create a new parameter (if --create flag is provided) + # 3) Update a parameter (if --update flag is provided) + if args.update: + if not args.parameter_id: + print("Error: --parameter-id is required for update operation") + return + + _print_header(f"Updating parameter: {args.parameter_id}") + + # First read the current parameter to show before state + current_param = client.policy_set_parameters.read( + args.policy_set_id, args.parameter_id + ) + print("Before update:") + print(f" Key: {current_param.key}") + value_display = ( + "***SENSITIVE***" if current_param.sensitive else current_param.value + ) + print(f" Value: {value_display}") + print(f" Sensitive: {current_param.sensitive}") + + # Update the parameter + update_options = PolicySetParameterUpdateOptions( + key=args.key if args.key else None, + value=args.value if args.value else None, + sensitive=args.sensitive if args.sensitive else None, + ) + + updated_param = client.policy_set_parameters.update( + args.policy_set_id, args.parameter_id, update_options + ) + + print("\nAfter update:") + print(f" Key: {updated_param.key}") + value_display = ( + "***SENSITIVE***" if updated_param.sensitive else updated_param.value + ) + print(f" Value: {value_display}") + print(f" Sensitive: {updated_param.sensitive}") + + # 4) Delete a parameter (if --delete flag is provided) + if args.delete: + if not args.parameter_id: + print("Error: --parameter-id is required for delete operation") + return + + _print_header(f"Deleting parameter: {args.parameter_id}") + + # First read the parameter to show what's being deleted + try: + param_to_delete = client.policy_set_parameters.read( + args.policy_set_id, args.parameter_id + ) + print("Parameter to delete:") + print(f" ID: {param_to_delete.id}") + print(f" Key: {param_to_delete.key}") + value_display = ( + "***SENSITIVE***" if param_to_delete.sensitive else param_to_delete.value + ) + print(f" Value: {value_display}") + print(f" Sensitive: {param_to_delete.sensitive}") + except Exception as e: + print(f"Error reading parameter: {e}") + return + + # Delete the parameter + client.policy_set_parameters.delete(args.policy_set_id, args.parameter_id) + print(f"\n✓ Successfully deleted parameter: {args.parameter_id}") + + # List remaining parameters + _print_header("Listing parameters after deletion") + remaining_list = client.policy_set_parameters.list(args.policy_set_id) + print(f"Total parameters: {remaining_list.total_count}") + if remaining_list.items: + for param in remaining_list.items: + value_display = "***SENSITIVE***" if param.sensitive else param.value + print(f"- {param.key}: {value_display} (sensitive={param.sensitive})") + else: + print("No parameters remaining.") + + # 5) Create a new parameter (if --create flag is provided) if args.create: + if not args.key: + print("Error: --key is required for create operation") + return + _print_header(f"Creating new parameter with key: {args.key}") create_options = PolicySetParameterCreateOptions( key=args.key, - value=args.value, + value=args.value if args.value else "", sensitive=args.sensitive, ) diff --git a/src/pytfe/resources/policy_set_parameter.py b/src/pytfe/resources/policy_set_parameter.py index 56fa9610..ab79afbf 100644 --- a/src/pytfe/resources/policy_set_parameter.py +++ b/src/pytfe/resources/policy_set_parameter.py @@ -12,6 +12,7 @@ PolicySetParameterCreateOptions, PolicySetParameterList, PolicySetParameterListOptions, + PolicySetParameterUpdateOptions, ) from ..models.variable import CategoryType from ..utils import valid_string, valid_string_id @@ -113,8 +114,6 @@ def read(self, policy_set_id: str, parameter_id: str) -> PolicySetParameter: ) return PolicySetParameter.model_validate(attrs) - -""" def update( self, policy_set_id: str, @@ -126,7 +125,27 @@ def update( if not valid_string_id(parameter_id): raise InvalidParamIDError() - return PolicySetParameter() + attributes = options.model_dump(by_alias=True, exclude_none=True) + payload = { + "data": { + "type": "vars", + "id": parameter_id, + "attributes": attributes, + } + } + r = self.t.request( + "PATCH", + path=f"api/v2/policy-sets/{policy_set_id}/parameters/{parameter_id}", + json_body=payload, + ) + jd = r.json() + data = jd.get("data", {}) + attrs = data.get("attributes", {}) + attrs["id"] = data.get("id") + attrs["policy_set"] = ( + data.get("relationships", {}).get("configurable", {}).get("data", {}) + ) + return PolicySetParameter.model_validate(attrs) def delete(self, policy_set_id: str, parameter_id: str) -> None: if not valid_string_id(policy_set_id): @@ -134,5 +153,8 @@ def delete(self, policy_set_id: str, parameter_id: str) -> None: if not valid_string_id(parameter_id): raise InvalidParamIDError() + self.t.request( + "DELETE", + path=f"api/v2/policy-sets/{policy_set_id}/parameters/{parameter_id}", + ) return None -""" From 2dfcd5da52542f1183e50f614d7a89c1a3b7e985 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Mon, 24 Nov 2025 17:06:10 +0530 Subject: [PATCH 3/6] feat(policy set parameter): adding iterator pattern to list method --- examples/policy_set_parameter.py | 69 ++++++++++++--------- src/pytfe/models/policy_set_parameter.py | 2 +- src/pytfe/resources/policy_set_parameter.py | 45 ++++++-------- 3 files changed, 59 insertions(+), 57 deletions(-) diff --git a/examples/policy_set_parameter.py b/examples/policy_set_parameter.py index 77302203..86b53633 100644 --- a/examples/policy_set_parameter.py +++ b/examples/policy_set_parameter.py @@ -26,13 +26,19 @@ def main(): ) parser.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) parser.add_argument("--policy-set-id", required=True, help="Policy Set ID") - parser.add_argument("--page", type=int, default=1) - parser.add_argument("--page-size", type=int, default=10) + parser.add_argument( + "--page-size", + type=int, + default=100, + help="Page size for fetching parameters (iterator fetches all pages)", + ) parser.add_argument("--create", action="store_true", help="Create a test parameter") parser.add_argument("--read", action="store_true", help="Read a specific parameter") parser.add_argument("--update", action="store_true", help="Update a parameter") parser.add_argument("--delete", action="store_true", help="Delete a parameter") - parser.add_argument("--parameter-id", help="Parameter ID for read/update/delete operation") + parser.add_argument( + "--parameter-id", help="Parameter ID for read/update/delete operation" + ) parser.add_argument("--key", help="Parameter key for creation/update") parser.add_argument("--value", help="Parameter value for creation/update") parser.add_argument( @@ -47,28 +53,25 @@ def main(): _print_header(f"Listing parameters for policy set: {args.policy_set_id}") options = PolicySetParameterListOptions( - page_number=args.page, page_size=args.page_size, ) - param_list = client.policy_set_parameters.list(args.policy_set_id, options) - - print(f"Total parameters: {param_list.total_count}") - print(f"Page {param_list.current_page} of {param_list.total_pages}") - print() + param_count = 0 + for param in client.policy_set_parameters.list(args.policy_set_id, options): + param_count += 1 + # Sensitive parameters will have masked values + value_display = "***SENSITIVE***" if param.sensitive else param.value + print(f"- {param.id}") + print(f" Key: {param.key}") + print(f" Value: {value_display}") + print(f" Category: {param.category.value}") + print(f" Sensitive: {param.sensitive}") + print() - if not param_list.items: + if param_count == 0: print("No parameters found.") else: - for param in param_list.items: - # Sensitive parameters will have masked values - value_display = "***SENSITIVE***" if param.sensitive else param.value - print(f"- {param.id}") - print(f" Key: {param.key}") - print(f" Value: {value_display}") - print(f" Category: {param.category.value}") - print(f" Sensitive: {param.sensitive}") - print() + print(f"Total: {param_count} parameters") # 2) Read a specific parameter (if --read flag is provided) if args.read: @@ -143,7 +146,9 @@ def main(): print(f" ID: {param_to_delete.id}") print(f" Key: {param_to_delete.key}") value_display = ( - "***SENSITIVE***" if param_to_delete.sensitive else param_to_delete.value + "***SENSITIVE***" + if param_to_delete.sensitive + else param_to_delete.value ) print(f" Value: {value_display}") print(f" Sensitive: {param_to_delete.sensitive}") @@ -157,14 +162,17 @@ def main(): # List remaining parameters _print_header("Listing parameters after deletion") - remaining_list = client.policy_set_parameters.list(args.policy_set_id) - print(f"Total parameters: {remaining_list.total_count}") - if remaining_list.items: - for param in remaining_list.items: - value_display = "***SENSITIVE***" if param.sensitive else param.value - print(f"- {param.key}: {value_display} (sensitive={param.sensitive})") - else: + print("Remaining parameters:") + remaining_count = 0 + for param in client.policy_set_parameters.list(args.policy_set_id): + remaining_count += 1 + value_display = "***SENSITIVE***" if param.sensitive else param.value + print(f"- {param.key}: {value_display} (sensitive={param.sensitive})") + + if remaining_count == 0: print("No parameters remaining.") + else: + print(f"\nTotal: {remaining_count} parameters") # 5) Create a new parameter (if --create flag is provided) if args.create: @@ -193,11 +201,12 @@ def main(): # List again to show the new parameter _print_header("Listing parameters after creation") - updated_list = client.policy_set_parameters.list(args.policy_set_id) - print(f"Total parameters: {updated_list.total_count}") - for param in updated_list.items: + param_count = 0 + for param in client.policy_set_parameters.list(args.policy_set_id): + param_count += 1 value_display = "***SENSITIVE***" if param.sensitive else param.value print(f"- {param.key}: {value_display} (sensitive={param.sensitive})") + print(f"\nTotal: {param_count} parameters") if __name__ == "__main__": diff --git a/src/pytfe/models/policy_set_parameter.py b/src/pytfe/models/policy_set_parameter.py index 8666a630..366b6df4 100644 --- a/src/pytfe/models/policy_set_parameter.py +++ b/src/pytfe/models/policy_set_parameter.py @@ -33,7 +33,7 @@ class PolicySetParameterList(BaseModel): class PolicySetParameterListOptions(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - page_number: int | None = Field(None, alias="page[number]") + # page_number: int | None = Field(None, alias="page[number]") page_size: int | None = Field(None, alias="page[size]") diff --git a/src/pytfe/resources/policy_set_parameter.py b/src/pytfe/resources/policy_set_parameter.py index ab79afbf..11378543 100644 --- a/src/pytfe/resources/policy_set_parameter.py +++ b/src/pytfe/resources/policy_set_parameter.py @@ -1,5 +1,8 @@ from __future__ import annotations +from collections.abc import Iterator +from typing import Any + from ..errors import ( InvalidCategoryError, InvalidParamIDError, @@ -10,7 +13,6 @@ from ..models.policy_set_parameter import ( PolicySetParameter, PolicySetParameterCreateOptions, - PolicySetParameterList, PolicySetParameterListOptions, PolicySetParameterUpdateOptions, ) @@ -19,6 +21,16 @@ from ._base import _Service +def _policy_set_parameter_from(d: dict[str, Any]) -> PolicySetParameter: + """Convert API response dict to PolicySetParameter model.""" + attrs = d.get("attributes", {}) + attrs["id"] = d.get("id") + attrs["policy_set"] = ( + d.get("relationships", {}).get("configurable", {}).get("data", {}) + ) + return PolicySetParameter.model_validate(attrs) + + class PolicySetParameters(_Service): """ PolicySetParameters describes all the parameter related methods that the Terraform Enterprise API supports. @@ -27,35 +39,14 @@ class PolicySetParameters(_Service): def list( self, policy_set_id: str, options: PolicySetParameterListOptions | None = None - ) -> PolicySetParameterList: + ) -> Iterator[PolicySetParameter]: """List all the parameters associated with the given policy-set.""" if not valid_string_id(policy_set_id): raise InvalidPolicySetIDError() params = options.model_dump(by_alias=True, exclude_none=True) if options else {} - r = self.t.request( - "GET", - path=f"api/v2/policy-sets/{policy_set_id}/parameters", - params=params, - ) - jd = r.json() - items = [] - meta = jd.get("meta", {}) - pagination = meta.get("pagination", {}) - for d in jd.get("data", []): - attrs = d.get("attributes", {}) - attrs["id"] = d.get("id") - attrs["policy_set"] = ( - d.get("relationships", {}).get("configurable", {}).get("data", {}) - ) - items.append(PolicySetParameter.model_validate(attrs)) - return PolicySetParameterList( - 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"), - ) + path = f"/api/v2/policy-sets/{policy_set_id}/parameters" + for item in self._list(path, params=params): + yield _policy_set_parameter_from(item) def create( self, policy_set_id: str, options: PolicySetParameterCreateOptions @@ -120,6 +111,7 @@ def update( parameter_id: str, options: PolicySetParameterUpdateOptions, ) -> PolicySetParameter: + """Update values of an existing parameter.""" if not valid_string_id(policy_set_id): raise InvalidPolicySetIDError() @@ -148,6 +140,7 @@ def update( return PolicySetParameter.model_validate(attrs) def delete(self, policy_set_id: str, parameter_id: str) -> None: + """Delete a parameter by its ID.""" if not valid_string_id(policy_set_id): raise InvalidPolicySetIDError() From 9db92e9d1c473bb43b622976cc191955ddd244e0 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Tue, 25 Nov 2025 14:47:46 +0530 Subject: [PATCH 4/6] test(policy set parameter): added unit tests --- tests/units/test_policy_set_parameter.py | 393 +++++++++++++++++++++++ 1 file changed, 393 insertions(+) create mode 100644 tests/units/test_policy_set_parameter.py diff --git a/tests/units/test_policy_set_parameter.py b/tests/units/test_policy_set_parameter.py new file mode 100644 index 00000000..05c2c4d1 --- /dev/null +++ b/tests/units/test_policy_set_parameter.py @@ -0,0 +1,393 @@ +"""Unit tests for the policy_set_parameter module.""" + +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import ( + InvalidCategoryError, + InvalidParamIDError, + InvalidPolicySetIDError, + RequiredKeyError, +) +from pytfe.models import ( + CategoryType, + PolicySetParameter, + PolicySetParameterCreateOptions, + PolicySetParameterListOptions, + PolicySetParameterUpdateOptions, +) +from pytfe.resources.policy_set_parameter import PolicySetParameters + + +class TestPolicySetParameters: + """Test the PolicySetParameters service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def policy_set_parameters_service(self, mock_transport): + """Create a PolicySetParameters service with mocked transport.""" + return PolicySetParameters(mock_transport) + + def test_list_parameters_validations(self, policy_set_parameters_service): + """Test list method with invalid policy set ID.""" + + # Test empty policy set ID + with pytest.raises(InvalidPolicySetIDError): + list(policy_set_parameters_service.list("")) + + # Test None policy set ID + with pytest.raises(InvalidPolicySetIDError): + list(policy_set_parameters_service.list(None)) + + def test_list_parameters_success_without_options( + self, policy_set_parameters_service + ): + """Test successful list operation without options.""" + + mock_data = [ + { + "id": "var-123", + "attributes": { + "key": "test_param", + "value": "test_value", + "category": "policy-set", + "sensitive": False, + }, + "relationships": { + "configurable": { + "data": {"id": "polset-123", "type": "policy-sets"} + } + }, + } + ] + + with patch.object(policy_set_parameters_service, "_list") as mock_list: + mock_list.return_value = iter(mock_data) + + result = list(policy_set_parameters_service.list("polset-123")) + + mock_list.assert_called_once_with( + "/api/v2/policy-sets/polset-123/parameters", params={} + ) + + assert len(result) == 1 + assert result[0].id == "var-123" + assert result[0].key == "test_param" + assert result[0].value == "test_value" + assert result[0].category == CategoryType.POLICY_SET + assert result[0].sensitive is False + + def test_list_parameters_with_options(self, policy_set_parameters_service): + """Test successful list operation with pagination options.""" + + mock_data = [] + + with patch.object(policy_set_parameters_service, "_list") as mock_list: + mock_list.return_value = iter(mock_data) + + options = PolicySetParameterListOptions(page_size=10) + result = list(policy_set_parameters_service.list("polset-123", options)) + + mock_list.assert_called_once_with( + "/api/v2/policy-sets/polset-123/parameters", + params={"page[size]": 10}, + ) + + assert len(result) == 0 + + def test_list_parameters_returns_iterator(self, policy_set_parameters_service): + """Test that list method returns an iterator.""" + + with patch.object(policy_set_parameters_service, "_list") as mock_list: + mock_list.return_value = iter([]) + + result = policy_set_parameters_service.list("polset-123") + + # Verify it's an iterator + assert hasattr(result, "__iter__") + assert hasattr(result, "__next__") + + def test_create_parameter_validations(self, policy_set_parameters_service): + """Test create method validations.""" + + # Test invalid policy set ID + options = PolicySetParameterCreateOptions(key="test") + with pytest.raises(InvalidPolicySetIDError): + policy_set_parameters_service.create("", options) + + # Test missing key + options = PolicySetParameterCreateOptions(key="") + with pytest.raises(RequiredKeyError): + policy_set_parameters_service.create("polset-123", options) + + # Test invalid category (not policy-set) + options = PolicySetParameterCreateOptions( + key="test", category=CategoryType.TERRAFORM + ) + with pytest.raises(InvalidCategoryError): + policy_set_parameters_service.create("polset-123", options) + + def test_create_parameter_success( + self, policy_set_parameters_service, mock_transport + ): + """Test successful create operation.""" + + mock_response_data = { + "data": { + "id": "var-456", + "attributes": { + "key": "new_param", + "value": "new_value", + "category": "policy-set", + "sensitive": False, + }, + "relationships": { + "configurable": { + "data": {"id": "polset-123", "type": "policy-sets"} + } + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + options = PolicySetParameterCreateOptions( + key="new_param", value="new_value", sensitive=False + ) + + result = policy_set_parameters_service.create("polset-123", options) + + mock_transport.request.assert_called_once_with( + "POST", + path="api/v2/policy-sets/polset-123/parameters", + json_body={ + "data": { + "type": "vars", + "attributes": { + "key": "new_param", + "value": "new_value", + "category": "policy-set", + "sensitive": False, + }, + } + }, + ) + + assert isinstance(result, PolicySetParameter) + assert result.id == "var-456" + assert result.key == "new_param" + assert result.value == "new_value" + + def test_create_sensitive_parameter( + self, policy_set_parameters_service, mock_transport + ): + """Test creating a sensitive parameter.""" + + mock_response_data = { + "data": { + "id": "var-789", + "attributes": { + "key": "secret_param", + "value": None, + "category": "policy-set", + "sensitive": True, + }, + "relationships": { + "configurable": { + "data": {"id": "polset-123", "type": "policy-sets"} + } + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + options = PolicySetParameterCreateOptions( + key="secret_param", value="secret_value", sensitive=True + ) + + result = policy_set_parameters_service.create("polset-123", options) + + assert isinstance(result, PolicySetParameter) + assert result.id == "var-789" + assert result.key == "secret_param" + assert result.value is None # Sensitive values are not returned + assert result.sensitive is True + + def test_read_parameter_validations(self, policy_set_parameters_service): + """Test read method validations.""" + + # Test invalid policy set ID + with pytest.raises(InvalidPolicySetIDError): + policy_set_parameters_service.read("", "var-123") + + # Test invalid parameter ID + with pytest.raises(InvalidParamIDError): + policy_set_parameters_service.read("polset-123", "") + + def test_read_parameter_success( + self, policy_set_parameters_service, mock_transport + ): + """Test successful read operation.""" + + mock_response_data = { + "data": { + "id": "var-789", + "attributes": { + "key": "existing_param", + "value": "existing_value", + "category": "policy-set", + "sensitive": False, + }, + "relationships": { + "configurable": { + "data": {"id": "polset-123", "type": "policy-sets"} + } + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + result = policy_set_parameters_service.read("polset-123", "var-789") + + mock_transport.request.assert_called_once_with( + "GET", path="api/v2/policy-sets/polset-123/parameters/var-789" + ) + + assert isinstance(result, PolicySetParameter) + assert result.id == "var-789" + assert result.key == "existing_param" + assert result.value == "existing_value" + + def test_update_parameter_validations(self, policy_set_parameters_service): + """Test update method validations.""" + + options = PolicySetParameterUpdateOptions(value="updated") + + # Test invalid policy set ID + with pytest.raises(InvalidPolicySetIDError): + policy_set_parameters_service.update("", "var-123", options) + + # Test invalid parameter ID + with pytest.raises(InvalidParamIDError): + policy_set_parameters_service.update("polset-123", "", options) + + def test_update_parameter_success( + self, policy_set_parameters_service, mock_transport + ): + """Test successful update operation.""" + + mock_response_data = { + "data": { + "id": "var-789", + "attributes": { + "key": "updated_param", + "value": "updated_value", + "category": "policy-set", + "sensitive": False, + }, + "relationships": { + "configurable": { + "data": {"id": "polset-123", "type": "policy-sets"} + } + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + options = PolicySetParameterUpdateOptions( + key="updated_param", value="updated_value" + ) + + result = policy_set_parameters_service.update("polset-123", "var-789", options) + + mock_transport.request.assert_called_once_with( + "PATCH", + path="api/v2/policy-sets/polset-123/parameters/var-789", + json_body={ + "data": { + "type": "vars", + "id": "var-789", + "attributes": {"key": "updated_param", "value": "updated_value"}, + } + }, + ) + + assert isinstance(result, PolicySetParameter) + assert result.id == "var-789" + assert result.key == "updated_param" + assert result.value == "updated_value" + + def test_update_parameter_to_sensitive( + self, policy_set_parameters_service, mock_transport + ): + """Test updating a parameter to make it sensitive.""" + + mock_response_data = { + "data": { + "id": "var-789", + "attributes": { + "key": "param", + "value": None, + "category": "policy-set", + "sensitive": True, + }, + "relationships": { + "configurable": { + "data": {"id": "polset-123", "type": "policy-sets"} + } + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + options = PolicySetParameterUpdateOptions(sensitive=True) + + result = policy_set_parameters_service.update("polset-123", "var-789", options) + + assert isinstance(result, PolicySetParameter) + assert result.sensitive is True + assert result.value is None + + def test_delete_parameter_validations(self, policy_set_parameters_service): + """Test delete method validations.""" + + # Test invalid policy set ID + with pytest.raises(InvalidPolicySetIDError): + policy_set_parameters_service.delete("", "var-123") + + # Test invalid parameter ID + with pytest.raises(InvalidParamIDError): + policy_set_parameters_service.delete("polset-123", "") + + def test_delete_parameter_success( + self, policy_set_parameters_service, mock_transport + ): + """Test successful delete operation.""" + + result = policy_set_parameters_service.delete("polset-123", "var-789") + + mock_transport.request.assert_called_once_with( + "DELETE", path="api/v2/policy-sets/polset-123/parameters/var-789" + ) + + assert result is None From e7a60334a3abf4325846969abea7d6195488f76c Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Wed, 26 Nov 2025 13:17:23 +0530 Subject: [PATCH 5/6] feat(policy set parameter): added a helper function --- examples/policy_set_parameter.py | 2 +- src/pytfe/models/__init__.py | 2 - src/pytfe/models/policy_set_parameter.py | 12 ----- src/pytfe/resources/policy_set_parameter.py | 51 +++++++-------------- 4 files changed, 17 insertions(+), 50 deletions(-) diff --git a/examples/policy_set_parameter.py b/examples/policy_set_parameter.py index 86b53633..c2ffea37 100644 --- a/examples/policy_set_parameter.py +++ b/examples/policy_set_parameter.py @@ -30,7 +30,7 @@ def main(): "--page-size", type=int, default=100, - help="Page size for fetching parameters (iterator fetches all pages)", + help="Page size for fetching parameters", ) parser.add_argument("--create", action="store_true", help="Create a test parameter") parser.add_argument("--read", action="store_true", help="Read a specific parameter") diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index b19c4464..c40737e0 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -136,7 +136,6 @@ from .policy_set_parameter import ( PolicySetParameter, PolicySetParameterCreateOptions, - PolicySetParameterList, PolicySetParameterListOptions, PolicySetParameterUpdateOptions, ) @@ -596,7 +595,6 @@ # Policy Set Parameters "PolicySetParameter", "PolicySetParameterCreateOptions", - "PolicySetParameterList", "PolicySetParameterListOptions", "PolicySetParameterUpdateOptions", "PolicyKind", diff --git a/src/pytfe/models/policy_set_parameter.py b/src/pytfe/models/policy_set_parameter.py index 366b6df4..01a88c25 100644 --- a/src/pytfe/models/policy_set_parameter.py +++ b/src/pytfe/models/policy_set_parameter.py @@ -19,21 +19,9 @@ class PolicySetParameter(BaseModel): policy_set: PolicySet = Field(..., alias="configurable") -class PolicySetParameterList(BaseModel): - model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - - items: list[PolicySetParameter] = Field(default_factory=list) - current_page: int | None = None - total_pages: int | None = None - prev_page: int | None = None - next_page: int | None = None - total_count: int | None = None - - class PolicySetParameterListOptions(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - # page_number: int | None = Field(None, alias="page[number]") page_size: int | None = Field(None, alias="page[size]") diff --git a/src/pytfe/resources/policy_set_parameter.py b/src/pytfe/resources/policy_set_parameter.py index 11378543..076579c4 100644 --- a/src/pytfe/resources/policy_set_parameter.py +++ b/src/pytfe/resources/policy_set_parameter.py @@ -21,16 +21,6 @@ from ._base import _Service -def _policy_set_parameter_from(d: dict[str, Any]) -> PolicySetParameter: - """Convert API response dict to PolicySetParameter model.""" - attrs = d.get("attributes", {}) - attrs["id"] = d.get("id") - attrs["policy_set"] = ( - d.get("relationships", {}).get("configurable", {}).get("data", {}) - ) - return PolicySetParameter.model_validate(attrs) - - class PolicySetParameters(_Service): """ PolicySetParameters describes all the parameter related methods that the Terraform Enterprise API supports. @@ -46,7 +36,7 @@ def list( params = options.model_dump(by_alias=True, exclude_none=True) if options else {} path = f"/api/v2/policy-sets/{policy_set_id}/parameters" for item in self._list(path, params=params): - yield _policy_set_parameter_from(item) + yield self._policy_set_parameter_from(item) def create( self, policy_set_id: str, options: PolicySetParameterCreateOptions @@ -75,14 +65,8 @@ def create( path=f"api/v2/policy-sets/{policy_set_id}/parameters", json_body=payload, ) - jd = r.json() - data = jd.get("data", {}) - attrs = data.get("attributes", {}) - attrs["id"] = data.get("id") - attrs["policy_set"] = ( - data.get("relationships", {}).get("configurable", {}).get("data", {}) - ) - return PolicySetParameter.model_validate(attrs) + data = r.json().get("data", {}) + return self._policy_set_parameter_from(data) def read(self, policy_set_id: str, parameter_id: str) -> PolicySetParameter: """Read a parameter by its ID.""" @@ -96,14 +80,8 @@ def read(self, policy_set_id: str, parameter_id: str) -> PolicySetParameter: "GET", path=f"api/v2/policy-sets/{policy_set_id}/parameters/{parameter_id}", ) - jd = r.json() - data = jd.get("data", {}) - attrs = data.get("attributes", {}) - attrs["id"] = data.get("id") - attrs["policy_set"] = ( - data.get("relationships", {}).get("configurable", {}).get("data", {}) - ) - return PolicySetParameter.model_validate(attrs) + data = r.json().get("data", {}) + return self._policy_set_parameter_from(data) def update( self, @@ -130,14 +108,8 @@ def update( path=f"api/v2/policy-sets/{policy_set_id}/parameters/{parameter_id}", json_body=payload, ) - jd = r.json() - data = jd.get("data", {}) - attrs = data.get("attributes", {}) - attrs["id"] = data.get("id") - attrs["policy_set"] = ( - data.get("relationships", {}).get("configurable", {}).get("data", {}) - ) - return PolicySetParameter.model_validate(attrs) + data = r.json().get("data", {}) + return self._policy_set_parameter_from(data) def delete(self, policy_set_id: str, parameter_id: str) -> None: """Delete a parameter by its ID.""" @@ -151,3 +123,12 @@ def delete(self, policy_set_id: str, parameter_id: str) -> None: path=f"api/v2/policy-sets/{policy_set_id}/parameters/{parameter_id}", ) return None + + def _policy_set_parameter_from(self, d: dict[str, Any]) -> PolicySetParameter: + """Convert API response dict to PolicySetParameter model.""" + attrs = d.get("attributes", {}) + attrs["id"] = d.get("id") + attrs["policy_set"] = ( + d.get("relationships", {}).get("configurable", {}).get("data", {}) + ) + return PolicySetParameter.model_validate(attrs) From 6d3f2d0ec4ff81f003cb81cd47196ed887d9fba9 Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Fri, 28 Nov 2025 14:44:46 +0530 Subject: [PATCH 6/6] feat(policy set parameters): modified the example --- examples/policy_set_parameter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/policy_set_parameter.py b/examples/policy_set_parameter.py index c2ffea37..9d6f6621 100644 --- a/examples/policy_set_parameter.py +++ b/examples/policy_set_parameter.py @@ -158,7 +158,7 @@ def main(): # Delete the parameter client.policy_set_parameters.delete(args.policy_set_id, args.parameter_id) - print(f"\n✓ Successfully deleted parameter: {args.parameter_id}") + print(f"\n Successfully deleted parameter: {args.parameter_id}") # List remaining parameters _print_header("Listing parameters after deletion")