From 81eeb0d818c13763b6b101ce09dd03d9da2bf23b Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Tue, 14 Oct 2025 12:49:21 +0530 Subject: [PATCH] Features providing Policy Evaluation, Policy Set Version and Policy Set Outcome API Specs --- examples/policy_evaluation.py | 107 ++++++++++++++++++++++ src/pytfe/client.py | 6 ++ src/pytfe/errors.py | 15 +++ src/pytfe/models/__init__.py | 17 ++++ src/pytfe/models/policy_evaluation.py | 94 +++++++++++++++++++ src/pytfe/models/policy_set_outcome.py | 66 +++++++++++++ src/pytfe/models/policy_set_version.py | 66 +++++++++++++ src/pytfe/resources/policy_evaluation.py | 53 +++++++++++ src/pytfe/resources/policy_set_outcome.py | 90 ++++++++++++++++++ src/pytfe/resources/policy_set_version.py | 82 +++++++++++++++++ 10 files changed, 596 insertions(+) create mode 100644 examples/policy_evaluation.py create mode 100644 src/pytfe/models/policy_evaluation.py create mode 100644 src/pytfe/models/policy_set_outcome.py create mode 100644 src/pytfe/resources/policy_evaluation.py create mode 100644 src/pytfe/resources/policy_set_outcome.py create mode 100644 src/pytfe/resources/policy_set_version.py diff --git a/examples/policy_evaluation.py b/examples/policy_evaluation.py new file mode 100644 index 0000000..9a0bd05 --- /dev/null +++ b/examples/policy_evaluation.py @@ -0,0 +1,107 @@ +from __future__ import annotations + +import argparse +import os + +from pytfe import TFEClient, TFEConfig +from pytfe.models import PolicyEvaluationListOptions + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser( + description="Policy Evaluations 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( + "--task-stage-id", + required=True, + help="Task stage ID to list policy evaluations for", + ) + parser.add_argument("--page", type=int, default=1) + parser.add_argument("--page-size", type=int, default=20) + args = parser.parse_args() + + if not args.token: + print("Error: TFE_TOKEN environment variable or --token argument is required") + return + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + # List all policy evaluations for the given task stage + _print_header(f"Listing policy evaluations for task stage: {args.task_stage_id}") + + options = PolicyEvaluationListOptions( + page_number=args.page, + page_size=args.page_size, + ) + + try: + pe_list = client.policy_evaluations.list(args.task_stage_id, options) + + print(f"Total policy evaluations: {pe_list.total_count}") + print(f"Page {pe_list.current_page} of {pe_list.total_pages}") + print() + + if not pe_list.items: + print("No policy evaluations found for this task stage.") + else: + for pe in pe_list.items: + print(f"- ID: {pe.id}") + print(f" Status: {pe.status}") + print(f" Policy Kind: {pe.policy_kind}") + + if pe.result_count: + print(" Result Count:") + if pe.result_count.passed is not None: + print(f" - Passed: {pe.result_count.passed}") + if pe.result_count.advisory_failed is not None: + print( + f" - Advisory Failed: {pe.result_count.advisory_failed}" + ) + if pe.result_count.mandatory_failed is not None: + print( + f" - Mandatory Failed: {pe.result_count.mandatory_failed}" + ) + if pe.result_count.errored is not None: + print(f" - Errored: {pe.result_count.errored}") + + if pe.status_timestamp: + print(" Status Timestamps:") + if pe.status_timestamp.passed_at: + print(f" - Passed At: {pe.status_timestamp.passed_at}") + if pe.status_timestamp.failed_at: + print(f" - Failed At: {pe.status_timestamp.failed_at}") + if pe.status_timestamp.running_at: + print(f" - Running At: {pe.status_timestamp.running_at}") + if pe.status_timestamp.canceled_at: + print(f" - Canceled At: {pe.status_timestamp.canceled_at}") + if pe.status_timestamp.errored_at: + print(f" - Errored At: {pe.status_timestamp.errored_at}") + + if pe.task_stage: + print(f" Task Stage: {pe.task_stage.id} ({pe.task_stage.type})") + + if pe.created_at: + print(f" Created At: {pe.created_at}") + if pe.updated_at: + print(f" Updated At: {pe.updated_at}") + + print() + + except Exception as e: + print(f"Error listing policy evaluations: {e}") + return + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 8af2a38..0ae6429 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -12,7 +12,10 @@ from .resources.plan import Plans from .resources.policy import Policies from .resources.policy_check import PolicyChecks +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_version import PolicySetVersions from .resources.projects import Projects from .resources.query_run import QueryRuns from .resources.registry_module import RegistryModules @@ -76,8 +79,11 @@ def __init__(self, config: TFEConfig | None = None): self.query_runs = QueryRuns(self._transport) self.run_events = RunEvents(self._transport) self.policies = Policies(self._transport) + self.policy_evaluations = PolicyEvaluations(self._transport) self.policy_checks = PolicyChecks(self._transport) self.policy_sets = PolicySets(self._transport) + self.policy_set_outcomes = PolicySetOutcomes(self._transport) + self.policy_set_versions = PolicySetVersions(self._transport) # SSH Keys self.ssh_keys = SSHKeys(self._transport) diff --git a/src/pytfe/errors.py b/src/pytfe/errors.py index 393e1d8..3eac2be 100644 --- a/src/pytfe/errors.py +++ b/src/pytfe/errors.py @@ -445,3 +445,18 @@ class InvalidPoliciesError(InvalidValues): def __init__(self, message: str = "must provide at least one policy"): super().__init__(message) + + +# Policy Evaluation errors +class InvalidTaskStageIDError(InvalidValues): + """Raised when an invalid task stage ID is provided.""" + + def __init__(self, message: str = "invalid value for task stage ID"): + super().__init__(message) + + +class InvalidPolicyEvaluationIDError(InvalidValues): + """Raised when an invalid policy evaluation ID is provided.""" + + def __init__(self, message: str = "invalid value for policy evaluation ID"): + super().__init__(message) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index cad12c1..a702bd2 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -115,6 +115,15 @@ PolicyStatus, PolicyStatusTimestamps, ) +from .policy_evaluation import ( + PolicyAttachable, + PolicyEvaluation, + PolicyEvaluationList, + PolicyEvaluationListOptions, + PolicyEvaluationStatus, + PolicyEvaluationStatusTimestamps, + PolicyResultCount, +) from .policy_set import ( PolicySet, PolicySetAddPoliciesOptions, @@ -526,6 +535,14 @@ "PolicyStatusTimestamps", "PolicyCheckListOptions", "PolicyCheckList", + # Policy Evaluation + "PolicyAttachable", + "PolicyEvaluation", + "PolicyEvaluationList", + "PolicyEvaluationListOptions", + "PolicyEvaluationStatus", + "PolicyEvaluationStatusTimestamps", + "PolicyResultCount", # Policy "Policy", "PolicyCreateOptions", diff --git a/src/pytfe/models/policy_evaluation.py b/src/pytfe/models/policy_evaluation.py new file mode 100644 index 0000000..86175e9 --- /dev/null +++ b/src/pytfe/models/policy_evaluation.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +from datetime import datetime +from enum import Enum + +from pydantic import BaseModel, ConfigDict, Field + +from .policy_types import PolicyKind + + +class PolicyEvaluationStatus(str, Enum): + """PolicyEvaluationStatus is an enum that represents all possible statuses for a policy evaluation""" + + POLICYEVALUATIONPASSED = "passed" + POLICYEVALUATIONFAILED = "failed" + POLICYEVALUATIONPENDING = "pending" + POLICYEVALUATIONRUNNING = "running" + POLICYEVALUATIONCANCELED = "canceled" + POLICYEVALUATIONERRORED = "errored" + POLICYEVALUATIONUNREACHABLE = "unreachable" + POLICYEVALUATIONOVERRIDDEN = "overridden" + + +class PolicyEvaluation(BaseModel): + """PolicyEvaluation represents the policy evaluations that are part of the task stage.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + status: PolicyEvaluationStatus | None = Field(None, alias="status") + policy_kind: PolicyKind | None = Field(None, alias="policy-kind") + status_timestamp: PolicyEvaluationStatusTimestamps | None = Field( + None, alias="status-timestamp" + ) + result_count: PolicyResultCount | None = Field(None, alias="result-count") + created_at: datetime | None = Field(None, alias="created-at") + updated_at: datetime | None = Field(None, alias="updated-at") + + # The task stage the policy evaluation belongs to + task_stage: PolicyAttachable | None = Field(None, alias="policy-attachable") + + +class PolicyEvaluationStatusTimestamps(BaseModel): + """PolicyEvaluationStatusTimestamps represents the set of timestamps recorded for a policy evaluation""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + passed_at: datetime | None = Field(None, alias="passed-at") + failed_at: datetime | None = Field(None, alias="failed-at") + running_at: datetime | None = Field(None, alias="running-at") + canceled_at: datetime | None = Field(None, alias="canceled-at") + errored_at: datetime | None = Field(None, alias="errored-at") + + +class PolicyAttachable(BaseModel): + """The task stage the policy evaluation belongs to""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + type: str | None = Field(None, alias="type") + + +class PolicyResultCount(BaseModel): + """PolicyResultCount represents the count of the policy results""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + advisory_failed: int | None = Field(None, alias="advisory-failed") + mandatory_failed: int | None = Field(None, alias="mandatory-failed") + passed: int | None = Field(None, alias="passed") + errored: int | None = Field(None, alias="errored") + + +class PolicyEvaluationList(BaseModel): + """PolicyEvaluationList represents a list of policy evaluations""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + items: list[PolicyEvaluation] | None = Field(default_factory=list) + current_page: int | None = None + next_page: str | None = None + prev_page: str | None = None + total_count: int | None = None + total_pages: int | None = None + + +class PolicyEvaluationListOptions(BaseModel): + """PolicyEvaluationListOptions represents the options for listing policy evaluations""" + + 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/models/policy_set_outcome.py b/src/pytfe/models/policy_set_outcome.py new file mode 100644 index 0000000..4059546 --- /dev/null +++ b/src/pytfe/models/policy_set_outcome.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from pydantic import BaseModel, ConfigDict, Field + +from .policy_evaluation import PolicyEvaluation, PolicyResultCount + + +class PolicySetOutcome(BaseModel): + """PolicySetOutcome represents outcome of the policy set that are part of the policy evaluation""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + outcomes: list[Outcome] = Field(default_factory=list, alias="outcomes") + error: str | None = Field(None, alias="error") + overridable: bool | None = Field(None, alias="overridable") + policy_set_name: str | None = Field(None, alias="policy-set-name") + policy_set_description: str | None = Field(None, alias="policy-set-description") + result_count: PolicyResultCount | None = Field(None, alias="result-count") + + # The policy evaluation that this outcome belongs to + policy_evaluation: PolicyEvaluation | None = Field(None, alias="policy-evaluation") + + +class Outcome(BaseModel): + """Outcome represents the outcome of the individual policy""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + enforcement_level: str | None = Field(None, alias="enforcement-level") + query: str | None = Field(None, alias="query") + status: str | None = Field(None, alias="status") + policy_name: str | None = Field(None, alias="policy-name") + description: str | None = Field(None, alias="description") + + +class PolicySetOutcomeList(BaseModel): + """PolicySetOutcomeList represents a list of policy set outcomes""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + items: list[PolicySetOutcome] | None = Field(default_factory=list) + current_page: int | None = None + next_page: str | None = None + prev_page: str | None = None + total_count: int | None = None + total_pages: int | None = None + + +class PolicySetOutcomeListFilter(BaseModel): + """PolicySetOutcomeListFilter represents the filters that are supported while listing a policy set outcome""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + status: str | None = Field(None, alias="status") + enforcement_level: str | None = Field(None, alias="enforcement-level") + + +class PolicySetOutcomeListOptions(BaseModel): + """PolicySetOutcomeListOptions represents the options for listing policy set outcomes.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + filter: dict[str, PolicySetOutcomeListFilter] | None = None + page_number: int | None = Field(None, alias="page[number]") + page_size: int | None = Field(None, alias="page[size]") diff --git a/src/pytfe/models/policy_set_version.py b/src/pytfe/models/policy_set_version.py index 8b8b0d7..4114856 100644 --- a/src/pytfe/models/policy_set_version.py +++ b/src/pytfe/models/policy_set_version.py @@ -1,10 +1,76 @@ from __future__ import annotations +from datetime import datetime +from enum import Enum +from typing import TYPE_CHECKING + from pydantic import BaseModel, ConfigDict, Field +if TYPE_CHECKING: + from .policy_set import PolicySet + + +class PolicySetVersionSource(str, Enum): + """ + PolicySetVersionSource represents a source type of a policy set version. + List all available sources for a Policy Set Version. + """ + + POLICY_SET_VERSION_SOURCE_API = "tfe-api" + POLICY_SET_VERSION_SOURCE_ADO = "ado" + POLICY_SET_VERSION_SOURCE_BITBUCKET = "bitbucket" + POLICY_SET_VERSION_SOURCE_GITHUB = "github" + POLICY_SET_VERSION_SOURCE_GITLAB = "gitlab" + + +class PolicySetVersionStatus(str, Enum): + """ + PolicySetVersionStatus represents a policy set version status. + List all available policy set version statuses. + """ + + POLICY_SET_VERSION_STATUS_ERRORED = "errored" + POLICY_SET_VERSION_STATUS_INGRESSING = "ingressing" + POLICY_SET_VERSION_STATUS_PENDING = "pending" + POLICY_SET_VERSION_STATUS_READY = "ready" + + +class PolicySetVersionStatusTimestamps(BaseModel): + """PolicySetVersionStatusTimestamps holds the timestamps for individual policy set version statuses.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + pending_at: datetime | None = Field(None, alias="pending-at") + ingressing_at: datetime | None = Field(None, alias="ingressing-at") + ready_at: datetime | None = Field(None, alias="ready-at") + errored_at: datetime | None = Field(None, alias="errored-at") + + +class PolicySetIngressAttributes(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + commit_sha: str | None = Field(None, alias="commit-sha") + commit_url: str | None = Field(None, alias="commit-url") + identifier: str | None = Field(None, alias="identifier") + class PolicySetVersion(BaseModel): + """PolicySetVersion represents a Terraform Enterprise Policy Set Version""" + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) id: str + source: PolicySetVersionSource | None = Field(None, alias="source") + status: PolicySetVersionStatus | None = Field(None, alias="status") + status_timestamps: PolicySetVersionStatusTimestamps | None = Field( + None, alias="status-timestamps" + ) + error_message: str | None = Field(None, alias="error-message") error: str | None = Field(None, alias="error") + created_at: datetime | None = Field(None, alias="created-at") + updated_at: datetime | None = Field(None, alias="updated-at") + ingress_attributes: PolicySetIngressAttributes | None = Field( + None, alias="ingress-attributes" + ) + policy_set: PolicySet | None = Field(None, alias="policy-set") + links: dict[str, str] | None = Field(None, alias="links") diff --git a/src/pytfe/resources/policy_evaluation.py b/src/pytfe/resources/policy_evaluation.py new file mode 100644 index 0000000..bc30193 --- /dev/null +++ b/src/pytfe/resources/policy_evaluation.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from ..errors import ( + InvalidTaskStageIDError, +) +from ..models.policy_evaluation import ( + PolicyEvaluation, + PolicyEvaluationList, + PolicyEvaluationListOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class PolicyEvaluations(_Service): + """ + PolicyEvalutations describes all the policy evaluation related methods that the Terraform Enterprise API supports. + TFE API docs: https://developer.hashicorp.com/terraform/cloud-docs/api-docs/policy-checks + """ + + def list( + self, task_stage_id: str, options: PolicyEvaluationListOptions | None = None + ) -> PolicyEvaluationList: + """ + **Note: This method is still in BETA and subject to change.** + List all policy evaluations in the task stage. Only available for OPA policies. + """ + if not valid_string_id(task_stage_id): + raise InvalidTaskStageIDError() + params = options.model_dump(by_alias=True) if options else {} + path = f"api/v2/task-stages/{task_stage_id}/policy-evaluations" + r = self.t.request("GET", path, params=params) + jd = r.json() + items = [] + meta = jd.get("meta", {}) + pagination = meta.get("pagination", {}) + for item in jd.get("data", []): + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + attrs["task-stage"] = ( + item.get("relationships", {}) + .get("policy-attachable", {}) + .get("data", {}) + ) + items.append(PolicyEvaluation.model_validate(attrs)) + return PolicyEvaluationList( + items=items, + current_page=pagination.get("current-page"), + next_page=pagination.get("next-page"), + prev_page=pagination.get("prev-page"), + total_count=pagination.get("total-count"), + total_pages=pagination.get("total-pages"), + ) diff --git a/src/pytfe/resources/policy_set_outcome.py b/src/pytfe/resources/policy_set_outcome.py new file mode 100644 index 0000000..56f7f34 --- /dev/null +++ b/src/pytfe/resources/policy_set_outcome.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +from ..errors import ( + InvalidPolicyEvaluationIDError, +) +from ..models.policy_set_outcome import ( + PolicySetOutcome, + PolicySetOutcomeList, + PolicySetOutcomeListOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class PolicySets(_Service): + """ + PolicySetOutcomes describes all the policy set outcome related methods that the Terraform Enterprise API supports. + TFE API docs: https://developer.hashicorp.com/terraform/cloud-docs/api-docs/policy-checks + """ + + def list( + self, + policy_evaluation_id: str, + options: PolicySetOutcomeListOptions | None = None, + ) -> PolicySetOutcomeList: + """ + **Note: This method is still in BETA and subject to change.** + List all policy set outcomes in the policy evaluation. Only available for OPA policies. + """ + if not valid_string_id(policy_evaluation_id): + raise InvalidPolicyEvaluationIDError() + + additional_query_params = self.build_query_string(options) + params = options.model_dump(by_alias=True) if options else {} + if additional_query_params: + params.update(additional_query_params) + path = f"api/v2/policy-evaluations/{policy_evaluation_id}/policy-set-outcomes" + r = self.t.request("GET", path, params=params) + jd = r.json() + items = [] + meta = jd.get("meta", {}) + pagination = meta.get("pagination", {}) + for item in jd.get("data", []): + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + attrs["policy-evaluation"] = ( + item.get("relationships", {}) + .get("policy-evaluation", {}) + .get("data", {}) + ) + items.append(PolicySetOutcome.model_validate(attrs)) + return PolicySetOutcomeList( + items=items, + current_page=pagination.get("current-page"), + next_page=pagination.get("next-page"), + prev_page=pagination.get("prev-page"), + total_count=pagination.get("total-count"), + total_pages=pagination.get("total-pages"), + ) + + def build_query_string( + self, options: PolicySetOutcomeListOptions | None + ) -> dict[str, str] | None: + """build_query_string takes the PolicySetOutcomeListOptions and returns a filters map.""" + result = {} + if options is None or options.filter is None: + return None + for key, value in options.filter.items(): + if value.status is not None: + result[f"filter[{key}][status]"] = value.status + if value.enforcement_level is not None: + result[f"filter[{key}][enforcement-level]"] = value.enforcement_level + return result + + def read(self, policy_set_outcome_id: str) -> PolicySetOutcome: + """ + **Note: This method is still in BETA and subject to change.** + Read a single policy set outcome by ID. Only available for OPA policies.""" + if not valid_string_id(policy_set_outcome_id): + raise InvalidPolicyEvaluationIDError() + path = f"api/v2/policy-set-outcomes/{policy_set_outcome_id}" + r = self.t.request("GET", path) + jd = r.json() + item = jd.get("data", {}) + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + attrs["policy-evaluation"] = ( + item.get("relationships", {}).get("policy-evaluation", {}).get("data", {}) + ) + return PolicySetOutcome.model_validate(attrs) diff --git a/src/pytfe/resources/policy_set_version.py b/src/pytfe/resources/policy_set_version.py new file mode 100644 index 0000000..b7c4234 --- /dev/null +++ b/src/pytfe/resources/policy_set_version.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from ..errors import ( + InvalidPolicySetIDError, +) +from ..models.policy_set_version import ( + PolicySetVersion, +) +from ..utils import pack_contents, valid_string_id +from ._base import _Service + + +class PolicySetVersions(_Service): + """ + PolicySetVersions describes all the Policy Set Version related methods that the Terraform Enterprise API supports. + TFE API docs: https://developer.hashicorp.com/terraform/cloud-docs/api-docs/policy-sets#create-a-policy-set-version + """ + + def create(self, policy_set_id: str) -> PolicySetVersion: + """Create is used to create a new Policy Set Version.""" + if not valid_string_id(policy_set_id): + raise InvalidPolicySetIDError() + r = self.t.request( + "POST", + f"/api/v2/policy-sets/{policy_set_id}/versions", + ) + jd = r.json() + attrs = jd.get("data", {}).get("attributes", {}) + attrs["id"] = jd.get("data", {}).get("id") + attrs["links"] = jd.get("data", {}).get("links", {}) + attrs["policy-set"] = ( + jd.get("data", {}) + .get("relationships", {}) + .get("policy-set", {}) + .get("data", {}) + ) + return PolicySetVersion.model_validate(attrs) + + def read(self, policy_set_version_id: str) -> PolicySetVersion: + """Read is used to read a Policy Set Version by its ID.""" + if not valid_string_id(policy_set_version_id): + raise InvalidPolicySetIDError() + r = self.t.request( + "GET", + f"/api/v2/policy-set-versions/{policy_set_version_id}", + ) + jd = r.json() + attrs = jd.get("data", {}).get("attributes", {}) + attrs["id"] = jd.get("data", {}).get("id") + attrs["links"] = jd.get("data", {}).get("links", {}) + attrs["policy-set"] = ( + jd.get("data", {}) + .get("relationships", {}) + .get("policy-set", {}) + .get("data", {}) + ) + return PolicySetVersion.model_validate(attrs) + + def upload(self, policy_set_version: PolicySetVersion, file_path: str) -> None: + """ + Upload uploads policy files. It takes a Policy Set Version and a path + to the set of sentinel files, which will be packaged by hashicorp/go-slug + before being uploaded. + """ + # Extract upload URL from policy set version links + if not policy_set_version.links or "upload" not in policy_set_version.links: + raise ValueError("the Policy Set Version does not contain an upload link") + + upload_url = policy_set_version.links["upload"] + if not upload_url: + raise ValueError("the Policy Set Version upload URL is empty") + + # Pack the policy files directory into a tar.gz archive + body = pack_contents(file_path) + + self.t.request( + "PUT", + upload_url, + data=body.getvalue(), + headers={"Content-Type": "application/octet-stream"}, + ) + return None