From 1b7880c03a5661c3eb97192da8260bd27d8c95dc Mon Sep 17 00:00:00 2001 From: Nimisha Shrivastava Date: Thu, 21 May 2026 15:10:50 +0530 Subject: [PATCH 1/5] feat(state-versions): add upload functionality support --- examples/state_versions.py | 242 ++++++++++++++++++++------ src/pytfe/models/state_version.py | 10 ++ src/pytfe/resources/state_versions.py | 74 +++++++- tests/units/test_state_version.py | 86 ++++++++- 4 files changed, 345 insertions(+), 67 deletions(-) diff --git a/examples/state_versions.py b/examples/state_versions.py index 61187619..8efdfd52 100644 --- a/examples/state_versions.py +++ b/examples/state_versions.py @@ -4,6 +4,8 @@ from __future__ import annotations import argparse +import hashlib +import json import os from pathlib import Path @@ -14,7 +16,9 @@ StateVersionCurrentOptions, StateVersionListOptions, StateVersionOutputsListOptions, + StateVersionReadOptions, ) +from pytfe.models.workspace import WorkspaceLockOptions def _print_header(title: str): @@ -23,6 +27,44 @@ def _print_header(title: str): print("=" * 80) +def _install_debug_hook(client: TFEClient, token: str) -> None: + """ + Wrap the transport's request() to print every URL and its headers. + The Authorization token value is masked so it is safe to share output. + """ + transport = client.state_versions.t + original_request = transport.request + + def _debug_request(method, path, **kwargs): + use_defaults = kwargs.get("use_default_headers", True) + extra_headers = kwargs.get("headers") or {} + + # Reconstruct exactly what the transport will send + if use_defaults: + sent_headers = dict(transport.headers) + sent_headers.update(extra_headers) + else: + sent_headers = dict(extra_headers) + + # Mask the bearer token so it is safe to print + display_headers = {} + for k, v in sent_headers.items(): + if k.lower() == "authorization": + masked = v[:14] + "***" + v[-4:] if len(v) > 18 else "***" + display_headers[k] = masked + else: + display_headers[k] = v + + url = transport._build_url(path) + print(f"\n [DEBUG] {method} {url}") + for k, v in display_headers.items(): + print(f" {k}: {v}") + + return original_request(method, path, **kwargs) + + transport.request = _debug_request + + def main(): parser = argparse.ArgumentParser( description="State Versions demo for python-tfe SDK" @@ -34,55 +76,80 @@ def main(): parser.add_argument("--org", required=True, help="Organization name") parser.add_argument("--workspace", required=True, help="Workspace name") parser.add_argument("--workspace-id", required=True, help="Workspace ID") - parser.add_argument("--download", help="Path to save downloaded current state") - parser.add_argument("--upload", help="Path to a .tfstate (or JSON state) to upload") + parser.add_argument( + "--download", help="Optional path to save downloaded current state" + ) + parser.add_argument( + "--upload", + help="Optional path to a .tfstate JSON to upload (defaults to current state with serial bumped by 1)", + ) + parser.add_argument( + "--skip-upload", + action="store_true", + help="Skip the upload demo (upload requires locking the workspace).", + ) + parser.add_argument( + "--demo-backing-data", + action="store_true", + help="Exercise TFE-only soft_delete/restore backing-data actions on the newly uploaded SV.", + ) parser.add_argument("--page-size", type=int, default=10) + parser.add_argument( + "--debug", + action="store_true", + help="Print every request URL and headers (token masked).", + ) args = parser.parse_args() cfg = TFEConfig(address=args.address, token=args.token) client = TFEClient(cfg) - options = StateVersionListOptions( - page_size=args.page_size, - organization=args.org, - workspace=args.workspace, - ) + if args.debug: + _install_debug_hook(client, args.token) - sv_list = list(client.state_versions.list(options)) - print(f"Total state versions: {len(sv_list)}") - print() - - for sv in sv_list: - print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") - - # 1) List all state versions across org and workspace filters - _print_header("Org-scoped listing via /api/v2/state-versions (first page)") - all_sv = client.state_versions.list( - StateVersionListOptions( - organization=args.org, workspace=args.workspace, page_size=args.page_size + # 1) List state versions filtered by org + workspace + _print_header("Listing state versions (filter[organization]+filter[workspace])") + sv_list = list( + client.state_versions.list( + StateVersionListOptions( + page_size=args.page_size, + organization=args.org, + workspace=args.workspace, + ) ) ) - for sv in all_sv: + print(f"Total state versions returned: {len(sv_list)}") + for sv in sv_list: print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") - # 2) Read the current state version (with outputs included if you want) - _print_header("Reading current state version") + # 2) Read the current state version with include=outputs + _print_header("read_current_with_options(include=outputs)") current = client.state_versions.read_current_with_options( args.workspace_id, StateVersionCurrentOptions(include=["outputs"]) ) - print( - f"Current SV: {current.id} status={current.status} durl={current.hosted_state_download_url}" + print(f"Current SV: {current.id} status={current.status}") + print(f" download_url: {current.hosted_state_download_url}") + print(f" json_download_url: {current.hosted_json_state_download_url}") + + # 3) Read by ID, with and without include options + _print_header("read(sv_id) and read_with_options(sv_id, include=[run,outputs])") + sv_read = client.state_versions.read(current.id) + print(f"read(): id={sv_read.id} serial={sv_read.serial}") + sv_read_opts = client.state_versions.read_with_options( + current.id, StateVersionReadOptions(include=["run", "outputs"]) ) + print(f"read_with_options(): id={sv_read_opts.id} serial={sv_read_opts.serial}") - # 3) (Optional) Download the current state (optional) + # 4) Download current state bytes + _print_header("download(current_sv_id)") + raw_current = client.state_versions.download(current.id) + print(f"Downloaded {len(raw_current)} bytes of state") if args.download: - _print_header(f"Downloading current state to: {args.download}") - raw = client.state_versions.download(current.id) - Path(args.download).write_bytes(raw) - print(f"Wrote {len(raw)} bytes to {args.download}") + Path(args.download).write_bytes(raw_current) + print(f" wrote bytes to {args.download}") - # 4) List outputs for the current state version (paged) - _print_header("Listing outputs (current state version)") + # 5) List outputs (by SV and via workspace shortcut) + _print_header("list_outputs(current_sv_id)") outs = list( client.state_versions.list_outputs( current.id, options=StateVersionOutputsListOptions(page_size=50) @@ -91,40 +158,101 @@ def main(): if not outs: print("No outputs found.") for o in outs: - # Sensitive outputs will have value = None print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") - if args.workspace_id: - # 4b) List outputs for the current state version via workspace endpoint - _print_header("Listing outputs via workspace endpoint") - outs2 = list( - client.state_version_outputs.read_current( - args.workspace_id, options=StateVersionOutputsListOptions(page_size=50) - ) + _print_header("state_version_outputs.read_current(workspace_id)") + outs2 = list( + client.state_version_outputs.read_current( + args.workspace_id, options=StateVersionOutputsListOptions(page_size=50) ) - if not outs2: - print("No outputs found.") - for o in outs2: - print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") + ) + if not outs2: + print("No outputs found.") + for o in outs2: + print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") + + # 6) Upload demo: requires the workspace to be locked. + if args.skip_upload: + _print_header("Skipping upload demo (--skip-upload)") + return - # 5) (Optional) Upload a new state file + _print_header("upload(workspace_id, raw_state=..., options=...)") if args.upload: - _print_header(f"Uploading new state from: {args.upload}") payload = Path(args.upload).read_bytes() + print(f"Using user-provided payload from {args.upload} ({len(payload)} bytes)") + else: + try: + state_obj = json.loads(raw_current.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as e: + print(f"Could not parse current state as JSON; skip upload: {e}") + return + state_obj["serial"] = int(state_obj.get("serial", 0)) + 1 + payload = json.dumps(state_obj).encode("utf-8") + print( + f"Synthesized payload from current state with serial bumped to " + f"{state_obj['serial']} ({len(payload)} bytes)" + ) + + try: + state_obj = json.loads(payload.decode("utf-8")) + serial = int(state_obj["serial"]) + lineage = state_obj.get("lineage") + except (KeyError, ValueError, json.JSONDecodeError) as e: + print(f"Upload input must be valid Terraform state JSON with a serial: {e}") + return + + md5 = hashlib.md5(payload).hexdigest() # nosec B324 + + locked = False + try: + client.workspaces.lock( + args.workspace_id, + WorkspaceLockOptions(reason="python-tfe state_versions example"), + ) + locked = True + print(f"Locked workspace {args.workspace_id}") + except Exception as e: + print(f"Could not lock workspace (continuing without lock): {e}") + + new_sv = None + try: + new_sv = client.state_versions.upload( + args.workspace_id, + raw_state=payload, + options=StateVersionCreateOptions( + serial=serial, + md5=md5, + lineage=lineage, + ), + ) + print( + f"Uploaded new SV: {new_sv.id} status={new_sv.status} serial={new_sv.serial}" + ) + except ErrStateVersionUploadNotSupported as e: + print(f"Upload not supported on this server: {e}") + except Exception as e: + print(f"Upload failed: {e}") + finally: + if locked: + try: + client.workspaces.unlock(args.workspace_id) + print(f"Unlocked workspace {args.workspace_id}") + except Exception as e: + print(f"Failed to unlock workspace: {e}") + + # 7) Optional: exercise TFE-only backing data actions on the new SV + if args.demo_backing_data and new_sv is not None: + _print_header("TFE-only backing data actions on the new SV") try: - # If your server supports signed uploads, this will: - # a) create SV (to get upload URL) - # b) PUT bytes to the signed URL - # c) read back the SV to return a hydrated object - new_sv = client.state_versions.upload( - args.workspace_id, - raw_state=payload, - options=StateVersionCreateOptions(), + client.state_versions.soft_delete_backing_data(new_sv.id) + print("soft_delete_backing_data: OK") + client.state_versions.restore_backing_data(new_sv.id) + print("restore_backing_data: OK") + print("(skipping permanently_delete_backing_data — irreversible)") + except Exception as e: + print( + f"Backing-data actions not available (likely HCP Terraform, not TFE): {e}" ) - print(f"Uploaded new SV: {new_sv.id} status={new_sv.status}") - except ErrStateVersionUploadNotSupported as e: - # Some older/self-hosted versions don’t support direct upload - print(f"Upload not supported on this server: {e}") if __name__ == "__main__": diff --git a/src/pytfe/models/state_version.py b/src/pytfe/models/state_version.py index dab42619..dbbe06f2 100644 --- a/src/pytfe/models/state_version.py +++ b/src/pytfe/models/state_version.py @@ -36,8 +36,18 @@ class StateVersion(BaseModel): hosted_state_download_url: str | None = Field( None, alias="hosted-state-download-url" ) + hosted_json_state_download_url: str | None = Field( + None, alias="hosted-json-state-download-url" + ) hosted_state_upload_url: str | None = Field(None, alias="hosted-state-upload-url") + hosted_json_state_upload_url: str | None = Field( + None, alias="hosted-json-state-upload-url" + ) status: StateVersionStatus | None = Field(None, alias="status") + serial: int | None = Field(None, alias="serial") + size: int | None = Field(None, alias="size") + terraform_version: str | None = Field(None, alias="terraform-version") + state_version: int | None = Field(None, alias="state-version") # Optional/advanced fields (present on newer servers; keep loose) resources_processed: bool | None = Field(None, alias="resources-processed") diff --git a/src/pytfe/resources/state_versions.py b/src/pytfe/resources/state_versions.py index e98e13bb..8ff07403 100644 --- a/src/pytfe/resources/state_versions.py +++ b/src/pytfe/resources/state_versions.py @@ -7,7 +7,9 @@ from typing import Any from urllib.parse import urlencode +from ..errors import ErrStateVersionUploadNotSupported from ..errors import NotFound +from ..errors import TFEError # Pydantic models for this feature from ..models.state_version import ( @@ -193,18 +195,66 @@ def create( **{k.replace("-", "_"): v for k, v in attr.items()}, ) - """ def upload( self, workspace: str, *, - raw_state: bytes | None = None, + raw_state: bytes | None, raw_json_state: bytes | None = None, - options: Optional[StateVersionCreateOptions] = None, - organization: Optional[str] = None, + options: StateVersionCreateOptions, + organization: str | None = None, ) -> StateVersion: - # TBD: Implements Upload State Functionality - """ + """ + Create a state version and upload state bytes to signed Archivist URLs. + + This mirrors Terraform's recommended workflow: + 1. POST /workspaces/:id/state-versions with serial+md5 and no inline state + 2. PUT raw state bytes to hosted-state-upload-url + 3. Optional PUT JSON state bytes to hosted-json-state-upload-url + 4. Read the state version again and return the refreshed object + """ + if raw_state is None: + raise ValueError("raw_state is required") + if options.state is not None or options.json_state is not None: + raise ValueError( + "options.state and options.json_state must be omitted when using upload" + ) + + try: + sv = self.create(workspace, options, organization=organization) + except TFEError as exc: + # Older servers can reject the create-without-inline-state flow. + if "param is missing or the value is empty: state" in str(exc): + raise ErrStateVersionUploadNotSupported( + "state version upload is not supported by this server" + ) from exc + raise + + if not sv.hosted_state_upload_url: + raise ErrStateVersionUploadNotSupported( + "hosted-state-upload-url not returned by server" + ) + + self.t.request( + "PUT", + sv.hosted_state_upload_url, + data=raw_state, + headers={"Content-Type": "application/octet-stream"}, + ) + + if raw_json_state is not None: + if not sv.hosted_json_state_upload_url: + raise ErrStateVersionUploadNotSupported( + "hosted-json-state-upload-url not returned by server" + ) + self.t.request( + "PUT", + sv.hosted_json_state_upload_url, + data=raw_json_state, + headers={"Content-Type": "application/octet-stream"}, + ) + + return self.read(sv.id) def download(self, state_version_id: str) -> bytes: """ @@ -226,9 +276,12 @@ def download(self, state_version_id: str) -> bytes: raise NotFound("download url not available for this state version") # Download the bytes from the signed Archivist URL (follow redirects). - # Avoid JSON:API headers here; Accept */* is fine. + # Avoid API default headers here; Accept */* is fine. resp = self.t.request( - "GET", url, allow_redirects=True, headers={"Accept": "application/json"} + "GET", + url, + allow_redirects=True, + headers={"Accept": "*/*"}, ) return resp.content @@ -244,7 +297,10 @@ def download_current(self, workspace_id: str) -> bytes: raise NotFound("download url not available for current state") resp = self.t.request( - "GET", url, allow_redirects=True, headers={"Accept": "*/*"} + "GET", + url, + allow_redirects=True, + headers={"Accept": "*/*"}, ) return resp.content diff --git a/tests/units/test_state_version.py b/tests/units/test_state_version.py index 11f67c8f..385fdd7e 100644 --- a/tests/units/test_state_version.py +++ b/tests/units/test_state_version.py @@ -5,7 +5,9 @@ import pytest from pytfe._http import HTTPTransport +from pytfe.errors import ErrStateVersionUploadNotSupported from pytfe.errors import NotFound +from pytfe.errors import TFEError from pytfe.models.state_version import ( StateVersion, StateVersionCreateOptions, @@ -131,6 +133,7 @@ def test_read_state_version_success(self, state_versions_service, mock_transport ) assert result.id == "sv-read-1" assert result.status == StateVersionStatus.FINALIZED + assert result.serial == 9 assert result.hosted_state_download_url == "https://example.com/download" def test_read_with_options_success(self, state_versions_service, mock_transport): @@ -204,6 +207,7 @@ def test_read_current_with_options_success( params={"include": "created_by"}, ) assert result.id == "sv-current-1" + assert result.serial == 9 def test_create_state_version_success(self, state_versions_service, mock_transport): """Test successful create() operation.""" @@ -247,6 +251,86 @@ def test_create_state_version_success(self, state_versions_service, mock_transpo assert result.id == "sv-new-1" assert result.status == StateVersionStatus.PENDING + def test_upload_state_version_success(self, state_versions_service, mock_transport): + """Test upload() creates, uploads raw bytes, and re-reads state version.""" + created_sv = StateVersion( + id="sv-upload-1", + created_at="2024-01-01T00:00:00Z", + status=StateVersionStatus.PENDING, + hosted_state_upload_url="https://example.com/upload-raw", + hosted_json_state_upload_url="https://example.com/upload-json", + ) + final_sv = StateVersion( + id="sv-upload-1", + created_at="2024-01-01T00:00:00Z", + status=StateVersionStatus.FINALIZED, + hosted_state_download_url="https://example.com/download-raw", + ) + options = StateVersionCreateOptions(serial=10, md5="abc123") + + with patch.object(state_versions_service, "create", return_value=created_sv): + with patch.object(state_versions_service, "read", return_value=final_sv): + result = state_versions_service.upload( + "ws-123", + raw_state=b"raw-state", + raw_json_state=b"json-state", + options=options, + ) + + assert result.id == "sv-upload-1" + assert result.status == StateVersionStatus.FINALIZED + assert mock_transport.request.call_count == 2 + mock_transport.request.assert_any_call( + "PUT", + "https://example.com/upload-raw", + data=b"raw-state", + headers={"Content-Type": "application/octet-stream"}, + ) + mock_transport.request.assert_any_call( + "PUT", + "https://example.com/upload-json", + data=b"json-state", + headers={"Content-Type": "application/octet-stream"}, + ) + + def test_upload_state_version_unsupported_on_create_error( + self, state_versions_service + ): + """Test upload() maps legacy create error text to typed unsupported error.""" + options = StateVersionCreateOptions(serial=10, md5="abc123") + legacy_err = TFEError("param is missing or the value is empty: state") + + with patch.object(state_versions_service, "create", side_effect=legacy_err): + with pytest.raises(ErrStateVersionUploadNotSupported): + state_versions_service.upload( + "ws-123", raw_state=b"raw-state", options=options + ) + + def test_upload_state_version_requires_signed_url(self, state_versions_service): + """Test upload() raises when server does not return hosted-state-upload-url.""" + created_sv = StateVersion( + id="sv-upload-2", + created_at="2024-01-01T00:00:00Z", + status=StateVersionStatus.PENDING, + hosted_state_upload_url=None, + ) + options = StateVersionCreateOptions(serial=10, md5="abc123") + + with patch.object(state_versions_service, "create", return_value=created_sv): + with pytest.raises(ErrStateVersionUploadNotSupported): + state_versions_service.upload( + "ws-123", raw_state=b"raw-state", options=options + ) + + def test_upload_state_version_rejects_inline_state(self, state_versions_service): + """Test upload() enforces omission of inline state/json-state in options.""" + options = StateVersionCreateOptions(serial=10, md5="abc123", state="abc") + + with pytest.raises(ValueError, match="must be omitted"): + state_versions_service.upload( + "ws-123", raw_state=b"raw-state", options=options + ) + def test_download_state_version_not_found_when_url_missing( self, state_versions_service ): @@ -283,7 +367,7 @@ def test_download_state_version_success( "GET", "https://example.com/signed-download", allow_redirects=True, - headers={"Accept": "application/json"}, + headers={"Accept": "*/*"}, ) assert result == b"{}" From 1a39cd08a6aedc227b1cf6ee945e32998212c7bd Mon Sep 17 00:00:00 2001 From: Nimisha Shrivastava Date: Thu, 21 May 2026 15:20:24 +0530 Subject: [PATCH 2/5] fixed lint --- src/pytfe/resources/state_versions.py | 6 +----- tests/units/test_state_version.py | 4 +--- 2 files changed, 2 insertions(+), 8 deletions(-) diff --git a/src/pytfe/resources/state_versions.py b/src/pytfe/resources/state_versions.py index 8ff07403..9a8d7bd3 100644 --- a/src/pytfe/resources/state_versions.py +++ b/src/pytfe/resources/state_versions.py @@ -7,11 +7,7 @@ from typing import Any from urllib.parse import urlencode -from ..errors import ErrStateVersionUploadNotSupported -from ..errors import NotFound -from ..errors import TFEError - -# Pydantic models for this feature +from ..errors import ErrStateVersionUploadNotSupported, NotFound, TFEError from ..models.state_version import ( StateVersion, StateVersionCreateOptions, diff --git a/tests/units/test_state_version.py b/tests/units/test_state_version.py index 385fdd7e..f9e3cd9b 100644 --- a/tests/units/test_state_version.py +++ b/tests/units/test_state_version.py @@ -5,9 +5,7 @@ import pytest from pytfe._http import HTTPTransport -from pytfe.errors import ErrStateVersionUploadNotSupported -from pytfe.errors import NotFound -from pytfe.errors import TFEError +from pytfe.errors import ErrStateVersionUploadNotSupported, NotFound, TFEError from pytfe.models.state_version import ( StateVersion, StateVersionCreateOptions, From a0a876f09eecc3ef242673247c70d9b51c803044 Mon Sep 17 00:00:00 2001 From: Nimisha Shrivastava Date: Fri, 22 May 2026 17:45:10 +0530 Subject: [PATCH 3/5] updated example files --- examples/state_versions.py | 265 ++++++++++++------------------------- 1 file changed, 85 insertions(+), 180 deletions(-) diff --git a/examples/state_versions.py b/examples/state_versions.py index 8efdfd52..3ff12a26 100644 --- a/examples/state_versions.py +++ b/examples/state_versions.py @@ -16,7 +16,6 @@ StateVersionCurrentOptions, StateVersionListOptions, StateVersionOutputsListOptions, - StateVersionReadOptions, ) from pytfe.models.workspace import WorkspaceLockOptions @@ -27,44 +26,6 @@ def _print_header(title: str): print("=" * 80) -def _install_debug_hook(client: TFEClient, token: str) -> None: - """ - Wrap the transport's request() to print every URL and its headers. - The Authorization token value is masked so it is safe to share output. - """ - transport = client.state_versions.t - original_request = transport.request - - def _debug_request(method, path, **kwargs): - use_defaults = kwargs.get("use_default_headers", True) - extra_headers = kwargs.get("headers") or {} - - # Reconstruct exactly what the transport will send - if use_defaults: - sent_headers = dict(transport.headers) - sent_headers.update(extra_headers) - else: - sent_headers = dict(extra_headers) - - # Mask the bearer token so it is safe to print - display_headers = {} - for k, v in sent_headers.items(): - if k.lower() == "authorization": - masked = v[:14] + "***" + v[-4:] if len(v) > 18 else "***" - display_headers[k] = masked - else: - display_headers[k] = v - - url = transport._build_url(path) - print(f"\n [DEBUG] {method} {url}") - for k, v in display_headers.items(): - print(f" {k}: {v}") - - return original_request(method, path, **kwargs) - - transport.request = _debug_request - - def main(): parser = argparse.ArgumentParser( description="State Versions demo for python-tfe SDK" @@ -76,80 +37,55 @@ def main(): parser.add_argument("--org", required=True, help="Organization name") parser.add_argument("--workspace", required=True, help="Workspace name") parser.add_argument("--workspace-id", required=True, help="Workspace ID") - parser.add_argument( - "--download", help="Optional path to save downloaded current state" - ) - parser.add_argument( - "--upload", - help="Optional path to a .tfstate JSON to upload (defaults to current state with serial bumped by 1)", - ) - parser.add_argument( - "--skip-upload", - action="store_true", - help="Skip the upload demo (upload requires locking the workspace).", - ) - parser.add_argument( - "--demo-backing-data", - action="store_true", - help="Exercise TFE-only soft_delete/restore backing-data actions on the newly uploaded SV.", - ) + parser.add_argument("--download", help="Path to save downloaded current state") + parser.add_argument("--upload", help="Path to a .tfstate (or JSON state) to upload") parser.add_argument("--page-size", type=int, default=10) - parser.add_argument( - "--debug", - action="store_true", - help="Print every request URL and headers (token masked).", - ) args = parser.parse_args() cfg = TFEConfig(address=args.address, token=args.token) client = TFEClient(cfg) - if args.debug: - _install_debug_hook(client, args.token) + options = StateVersionListOptions( + page_size=args.page_size, + organization=args.org, + workspace=args.workspace, + ) - # 1) List state versions filtered by org + workspace - _print_header("Listing state versions (filter[organization]+filter[workspace])") - sv_list = list( - client.state_versions.list( - StateVersionListOptions( - page_size=args.page_size, - organization=args.org, - workspace=args.workspace, - ) + sv_list = list(client.state_versions.list(options)) + print(f"Total state versions: {len(sv_list)}") + print() + + for sv in sv_list: + print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") + + # 1) List all state versions across org and workspace filters + _print_header("Org-scoped listing via /api/v2/state-versions (first page)") + all_sv = client.state_versions.list( + StateVersionListOptions( + organization=args.org, workspace=args.workspace, page_size=args.page_size ) ) - print(f"Total state versions returned: {len(sv_list)}") - for sv in sv_list: + for sv in all_sv: print(f"- {sv.id} | status={sv.status} | created_at={sv.created_at}") - # 2) Read the current state version with include=outputs - _print_header("read_current_with_options(include=outputs)") + # 2) Read the current state version (with outputs included if you want) + _print_header("Reading current state version") current = client.state_versions.read_current_with_options( args.workspace_id, StateVersionCurrentOptions(include=["outputs"]) ) - print(f"Current SV: {current.id} status={current.status}") - print(f" download_url: {current.hosted_state_download_url}") - print(f" json_download_url: {current.hosted_json_state_download_url}") - - # 3) Read by ID, with and without include options - _print_header("read(sv_id) and read_with_options(sv_id, include=[run,outputs])") - sv_read = client.state_versions.read(current.id) - print(f"read(): id={sv_read.id} serial={sv_read.serial}") - sv_read_opts = client.state_versions.read_with_options( - current.id, StateVersionReadOptions(include=["run", "outputs"]) + print( + f"Current SV: {current.id} status={current.status} durl={current.hosted_state_download_url}" ) - print(f"read_with_options(): id={sv_read_opts.id} serial={sv_read_opts.serial}") - # 4) Download current state bytes - _print_header("download(current_sv_id)") - raw_current = client.state_versions.download(current.id) - print(f"Downloaded {len(raw_current)} bytes of state") + # 3) (Optional) Download the current state (optional) if args.download: - Path(args.download).write_bytes(raw_current) - print(f" wrote bytes to {args.download}") + _print_header(f"Downloading current state to: {args.download}") + raw = client.state_versions.download(current.id) + Path(args.download).write_bytes(raw) + print(f"Wrote {len(raw)} bytes to {args.download}") - # 5) List outputs (by SV and via workspace shortcut) - _print_header("list_outputs(current_sv_id)") + # 4) List outputs for the current state version (paged) + _print_header("Listing outputs (current state version)") outs = list( client.state_versions.list_outputs( current.id, options=StateVersionOutputsListOptions(page_size=50) @@ -158,101 +94,70 @@ def main(): if not outs: print("No outputs found.") for o in outs: + # Sensitive outputs will have value = None print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") - _print_header("state_version_outputs.read_current(workspace_id)") - outs2 = list( - client.state_version_outputs.read_current( - args.workspace_id, options=StateVersionOutputsListOptions(page_size=50) + if args.workspace_id: + # 4b) List outputs for the current state version via workspace endpoint + _print_header("Listing outputs via workspace endpoint") + outs2 = list( + client.state_version_outputs.read_current( + args.workspace_id, options=StateVersionOutputsListOptions(page_size=50) + ) ) - ) - if not outs2: - print("No outputs found.") - for o in outs2: - print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") - - # 6) Upload demo: requires the workspace to be locked. - if args.skip_upload: - _print_header("Skipping upload demo (--skip-upload)") - return + if not outs2: + print("No outputs found.") + for o in outs2: + print(f"- {o.name}: sensitive={o.sensitive} type={o.type} value={o.value}") - _print_header("upload(workspace_id, raw_state=..., options=...)") + # 5) (Optional) Upload a new state file if args.upload: - payload = Path(args.upload).read_bytes() - print(f"Using user-provided payload from {args.upload} ({len(payload)} bytes)") - else: + _print_header(f"Uploading new state from: {args.upload}") try: - state_obj = json.loads(raw_current.decode("utf-8")) - except (UnicodeDecodeError, json.JSONDecodeError) as e: - print(f"Could not parse current state as JSON; skip upload: {e}") - return - state_obj["serial"] = int(state_obj.get("serial", 0)) + 1 - payload = json.dumps(state_obj).encode("utf-8") - print( - f"Synthesized payload from current state with serial bumped to " - f"{state_obj['serial']} ({len(payload)} bytes)" - ) - - try: - state_obj = json.loads(payload.decode("utf-8")) - serial = int(state_obj["serial"]) - lineage = state_obj.get("lineage") - except (KeyError, ValueError, json.JSONDecodeError) as e: - print(f"Upload input must be valid Terraform state JSON with a serial: {e}") - return - - md5 = hashlib.md5(payload).hexdigest() # nosec B324 - - locked = False - try: - client.workspaces.lock( - args.workspace_id, - WorkspaceLockOptions(reason="python-tfe state_versions example"), - ) - locked = True - print(f"Locked workspace {args.workspace_id}") - except Exception as e: - print(f"Could not lock workspace (continuing without lock): {e}") + payload = Path(args.upload).read_bytes() + state_obj = json.loads(payload.decode("utf-8")) + serial = int(state_obj["serial"]) + lineage = state_obj.get("lineage") + md5 = hashlib.md5(payload).hexdigest() # nosec B324 + locked_workspace = False - new_sv = None - try: - new_sv = client.state_versions.upload( - args.workspace_id, - raw_state=payload, - options=StateVersionCreateOptions( - serial=serial, - md5=md5, - lineage=lineage, - ), - ) - print( - f"Uploaded new SV: {new_sv.id} status={new_sv.status} serial={new_sv.serial}" - ) - except ErrStateVersionUploadNotSupported as e: - print(f"Upload not supported on this server: {e}") - except Exception as e: - print(f"Upload failed: {e}") - finally: - if locked: try: - client.workspaces.unlock(args.workspace_id) - print(f"Unlocked workspace {args.workspace_id}") - except Exception as e: - print(f"Failed to unlock workspace: {e}") - - # 7) Optional: exercise TFE-only backing data actions on the new SV - if args.demo_backing_data and new_sv is not None: - _print_header("TFE-only backing data actions on the new SV") - try: - client.state_versions.soft_delete_backing_data(new_sv.id) - print("soft_delete_backing_data: OK") - client.state_versions.restore_backing_data(new_sv.id) - print("restore_backing_data: OK") - print("(skipping permanently_delete_backing_data — irreversible)") - except Exception as e: + client.workspaces.lock( + args.workspace_id, + WorkspaceLockOptions(reason="python-tfe state_versions upload example"), + ) + locked_workspace = True + except Exception: + # Continue in case the workspace is already locked by the caller. + pass + + # If your server supports signed uploads, this will: + # a) create SV (to get upload URL) + # b) PUT bytes to the signed URL + # c) read back the SV to return a hydrated object + try: + new_sv = client.state_versions.upload( + args.workspace_id, + raw_state=payload, + options=StateVersionCreateOptions( + serial=serial, + md5=md5, + lineage=lineage, + ), + ) + finally: + if locked_workspace: + client.workspaces.unlock(args.workspace_id) + print(f"Uploaded new SV: {new_sv.id} status={new_sv.status}") + except FileNotFoundError: + print(f"Upload file not found: {args.upload}") + except (KeyError, ValueError, json.JSONDecodeError): print( - f"Backing-data actions not available (likely HCP Terraform, not TFE): {e}" + "Upload input must be a valid Terraform state JSON containing at least a serial value." ) + except ErrStateVersionUploadNotSupported as e: + # Some older/self-hosted versions don’t support direct upload + print(f"Upload not supported on this server: {e}") if __name__ == "__main__": From ebf72f37cc9c7d2aba3262e2fc8e8166ad6131d8 Mon Sep 17 00:00:00 2001 From: Nimisha Shrivastava Date: Fri, 22 May 2026 17:48:16 +0530 Subject: [PATCH 4/5] fixed formatting --- examples/state_versions.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/examples/state_versions.py b/examples/state_versions.py index 3ff12a26..e9a98d3d 100644 --- a/examples/state_versions.py +++ b/examples/state_versions.py @@ -124,7 +124,9 @@ def main(): try: client.workspaces.lock( args.workspace_id, - WorkspaceLockOptions(reason="python-tfe state_versions upload example"), + WorkspaceLockOptions( + reason="python-tfe state_versions upload example" + ), ) locked_workspace = True except Exception: From d3a730463b81fd6f8e4fa9a9ce84b137622bb8bb Mon Sep 17 00:00:00 2001 From: Nimisha Shrivastava Date: Fri, 22 May 2026 18:17:55 +0530 Subject: [PATCH 5/5] fixed download --- src/pytfe/resources/state_versions.py | 12 +++--------- tests/units/test_state_version.py | 2 +- 2 files changed, 4 insertions(+), 10 deletions(-) diff --git a/src/pytfe/resources/state_versions.py b/src/pytfe/resources/state_versions.py index 9a8d7bd3..c1014dae 100644 --- a/src/pytfe/resources/state_versions.py +++ b/src/pytfe/resources/state_versions.py @@ -272,12 +272,9 @@ def download(self, state_version_id: str) -> bytes: raise NotFound("download url not available for this state version") # Download the bytes from the signed Archivist URL (follow redirects). - # Avoid API default headers here; Accept */* is fine. + # Avoid JSON:API headers here; Accept */* is fine. resp = self.t.request( - "GET", - url, - allow_redirects=True, - headers={"Accept": "*/*"}, + "GET", url, allow_redirects=True, headers={"Accept": "application/json"} ) return resp.content @@ -293,10 +290,7 @@ def download_current(self, workspace_id: str) -> bytes: raise NotFound("download url not available for current state") resp = self.t.request( - "GET", - url, - allow_redirects=True, - headers={"Accept": "*/*"}, + "GET", url, allow_redirects=True, headers={"Accept": "*/*"} ) return resp.content diff --git a/tests/units/test_state_version.py b/tests/units/test_state_version.py index f9e3cd9b..e8717bb2 100644 --- a/tests/units/test_state_version.py +++ b/tests/units/test_state_version.py @@ -365,7 +365,7 @@ def test_download_state_version_success( "GET", "https://example.com/signed-download", allow_redirects=True, - headers={"Accept": "*/*"}, + headers={"Accept": "application/json"}, ) assert result == b"{}"