diff --git a/CHANGELOG.md b/CHANGELOG.md index e4aed710..14b39711 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # Unreleased +## Enhancements + +### New resources +* Added `client.stack_deployments` — list the deployments that belong to a stack. `list(stack_id, options=None)` (`GET /stacks/{stack_id}/stack-deployments`) returns an `Iterator[StackDeployment]`, with optional pagination (`page_size`) and `?include=` (`latest_deployment_run`, `latest_deployment_run.stack_configuration`) via `StackDeploymentListOptions`. The `stack` relationship is hydrated as a typed field; the `latest-deployment-run` relation is reachable via the lossless raw accessors (`deployment.related("latest-deployment-run")`). New models: `StackDeployment`, `StackDeploymentListOptions`, `StackDeploymentIncludeOpt`. + # Released # v1.2.0 diff --git a/examples/stack_configuration.py b/examples/stack_configuration.py index d51ace60..1d9c96d6 100644 --- a/examples/stack_configuration.py +++ b/examples/stack_configuration.py @@ -10,6 +10,7 @@ from pytfe.models import ( StackConfigurationCreateOptions, StackConfigurationListOptions, + StackConfigurationSource, ) @@ -82,7 +83,7 @@ def main(): else: print(f"Total: {config_count} stack configurations") - # 2) Create a new stack configuration + # 2) Create a new stack configuration (manual archive upload) if args.create: _print_header("Creating a new stack configuration") create_opts = StackConfigurationCreateOptions( @@ -97,7 +98,19 @@ def main(): print(f" Sequence: {config.sequence_number}") print(f" Created: {config.created_at}") - # 3) Read a specific stack configuration + # 3) Fetch latest config from VCS (stack must have a vcs_repo wired) + if args.fetch_from_vcs: + _print_header("Fetching latest configuration from VCS") + config = client.stack_configurations.create( + stack_id=args.stack_id, + source=StackConfigurationSource.FETCH, + ) + print(f"Triggered VCS fetch: {config.id}") + print(f" Status: {config.status.value if config.status else None}") + print(f" Sequence: {config.sequence_number}") + print(f" Created: {config.created_at}") + + # 4) Read a specific stack configuration if args.read: if not args.id: print("--id is required for --read") diff --git a/examples/stack_deployment.py b/examples/stack_deployment.py new file mode 100644 index 00000000..5d3f1333 --- /dev/null +++ b/examples/stack_deployment.py @@ -0,0 +1,84 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +""" +Terraform Cloud/Enterprise Stack Deployments Example + +Lists the deployments that belong to a Stack +(GET /stacks/:stack_id/stack-deployments). + +Prerequisites: + - Set TFE_TOKEN environment variable with your Terraform Cloud API token + - A Stack id (e.g. st-xxxxxxxx) + +Usage: + python examples/stack_deployment.py --stack-id st-xxxxxxxx + python examples/stack_deployment.py --stack-id st-xxxxxxxx --page-size 50 + python examples/stack_deployment.py --stack-id st-xxxxxxxx --include-latest-run +""" + +from __future__ import annotations + +import argparse +import os + +from pytfe import TFEClient, TFEConfig +from pytfe.models import ( + StackDeploymentIncludeOpt, + StackDeploymentListOptions, +) + + +def _print_header(title: str) -> None: + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main() -> int: + parser = argparse.ArgumentParser( + description="Stack Deployments 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("--stack-id", required=True, help="Stack ID (e.g. st-xxxxxxxx)") + parser.add_argument( + "--page-size", type=int, default=20, help="Page size for listing deployments" + ) + parser.add_argument( + "--include-latest-run", + action="store_true", + help="Request the latest_deployment_run related resource", + ) + args = parser.parse_args() + + if not args.token: + print("TFE_TOKEN is not set") + return 2 + + client = TFEClient(TFEConfig(address=args.address, token=args.token)) + + include = ( + [StackDeploymentIncludeOpt.LATEST_DEPLOYMENT_RUN] + if args.include_latest_run + else None + ) + options = StackDeploymentListOptions(page_size=args.page_size, include=include) + + _print_header(f"Listing deployments for stack {args.stack_id}") + for deployment in client.stack_deployments.list(args.stack_id, options): + print(f" - {deployment.id} {deployment.name}") + if deployment.stack: + print(f" stack: {deployment.stack.id}") + # The latest-deployment-run relation is not modelled as a typed field, + # but it is always reachable losslessly via the raw escape hatch. + for ref in deployment.related("latest-deployment-run"): + print(f" latest-deployment-run: {ref.get('id')}") + + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 14536448..63d711fe 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -60,6 +60,7 @@ from .resources.ssh_keys import SSHKeys from .resources.stack import Stacks from .resources.stack_configuration import StackConfigurations +from .resources.stack_deployment import StackDeployments from .resources.state_version_outputs import StateVersionOutputs from .resources.state_versions import StateVersions from .resources.subscription import Subscriptions @@ -206,6 +207,7 @@ def __init__(self, config: TFEConfig | None = None): # Stack resources self.stacks = Stacks(self._transport) self.stack_configurations = StackConfigurations(self._transport) + self.stack_deployments = StackDeployments(self._transport) # State and execution resources self.state_versions = StateVersions(self._transport) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 0fb9efc1..8e49e95c 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -544,6 +544,11 @@ StackConfigurationSource, StackConfigurationStatus, ) +from .stack_deployment import ( + StackDeployment, + StackDeploymentIncludeOpt, + StackDeploymentListOptions, +) from .state_version import ( StateVersion, StateVersionCreateOptions, @@ -890,6 +895,10 @@ "StackConfigurationReadOptions", "StackConfigurationSource", "StackConfigurationStatus", + # Stack Deployment + "StackDeployment", + "StackDeploymentIncludeOpt", + "StackDeploymentListOptions", # Query runs "QueryRun", "QueryRunActions", diff --git a/src/pytfe/models/stack_deployment.py b/src/pytfe/models/stack_deployment.py new file mode 100644 index 00000000..67da03c7 --- /dev/null +++ b/src/pytfe/models/stack_deployment.py @@ -0,0 +1,46 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from enum import Enum + +from pydantic import BaseModel, ConfigDict, Field + +from ._base import TFEModel +from .stack import Stack + + +class StackDeploymentIncludeOpt(str, Enum): + """StackDeploymentIncludeOpt represents include options for stack deployment endpoints.""" + + LATEST_DEPLOYMENT_RUN = "latest_deployment_run" + LATEST_DEPLOYMENT_RUN_STACK_CONFIGURATION = ( + "latest_deployment_run.stack_configuration" + ) + + +class StackDeployment(TFEModel): + """StackDeployment represents a deployment that belongs to a stack. + + JSON:API type: ``stack-deployments``. + """ + + model_config = ConfigDict( + populate_by_name=True, validate_by_name=True, extra="allow" + ) + + id: str + name: str | None = Field(default=None, alias="name") + + # Relations + stack: Stack | None = Field(default=None, alias="stack") + + +class StackDeploymentListOptions(BaseModel): + """StackDeploymentListOptions represents the options for listing stack deployments.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + page_size: int | None = Field(default=None, alias="page[size]") + include: list[StackDeploymentIncludeOpt] | None = None diff --git a/src/pytfe/resources/stack_deployment.py b/src/pytfe/resources/stack_deployment.py new file mode 100644 index 00000000..aa0c60ed --- /dev/null +++ b/src/pytfe/resources/stack_deployment.py @@ -0,0 +1,75 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +import builtins +from collections.abc import Iterator +from typing import Any + +from .._jsonapi import attach_jsonapi, parse_relationships +from ..errors import InvalidStackIDError +from ..models.stack import Stack +from ..models.stack_deployment import ( + StackDeployment, + StackDeploymentListOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class StackDeployments(_Service): + """Service for reading the deployments that belong to a stack.""" + + def list( + self, + stack_id: str, + options: StackDeploymentListOptions | None = None, + ) -> Iterator[StackDeployment]: + """List the deployments that belong to a stack. + + Args: + stack_id: The stack ID (e.g. ``"st-xxxxxxxx"``). + options: Optional pagination and includes, as a + :class:`StackDeploymentListOptions`. + + Returns: + A single-use ``Iterator[StackDeployment]``. Wrap with ``list(...)`` to + materialize the results or iterate more than once. + + Raises: + InvalidStackIDError: If ``stack_id`` is empty or malformed. + TFEError: If the API request fails. + + Example: + >>> for deployment in client.stack_deployments.list("st-xyz789"): + ... print(deployment.id, deployment.name) + """ + if not valid_string_id(stack_id): + raise InvalidStackIDError() + path = f"/api/v2/stacks/{stack_id}/stack-deployments" + params: dict[str, Any] = {} + if options: + if options.page_size is not None: + params["page[size]"] = options.page_size + if options.include: + params["include"] = ",".join([i.value for i in options.include]) + for item in self._list(path=path, params=params): + yield self._stack_deployment_from(item) + + def _stack_deployment_from( + self, + data: dict[str, Any], + included: builtins.list[dict[str, Any]] | None = None, + ) -> StackDeployment: + """Parse a StackDeployment from API response data.""" + attrs = dict(data.get("attributes", {})) + attrs["id"] = data.get("id") + attrs.update( + parse_relationships( + data.get("relationships"), + {"stack": Stack}, + included=included, + ) + ) + return attach_jsonapi(StackDeployment.model_validate(attrs), data, included) diff --git a/tests/units/test_stack_deployment.py b/tests/units/test_stack_deployment.py new file mode 100644 index 00000000..9116f01c --- /dev/null +++ b/tests/units/test_stack_deployment.py @@ -0,0 +1,149 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +"""Unit tests for the stack_deployment module.""" + +from unittest.mock import Mock + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import InvalidStackIDError +from pytfe.models.stack import Stack +from pytfe.models.stack_deployment import ( + StackDeployment, + StackDeploymentIncludeOpt, + StackDeploymentListOptions, +) +from pytfe.resources.stack_deployment import StackDeployments + + +class TestStackDeployments: + """Test the StackDeployments service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def service(self, mock_transport): + """Create a StackDeployments service with mocked transport.""" + return StackDeployments(mock_transport) + + @pytest.fixture + def stack_deployment_api_data(self): + """Typical API response item for a single stack deployment.""" + return { + "id": "st-MWvJsvy1FCg3bnXY-std-simple", + "type": "stack-deployments", + "attributes": {"name": "simple"}, + "relationships": { + "stack": {"data": {"id": "st-MWvJsvy1FCg3bnXY", "type": "stacks"}}, + "latest-deployment-run": { + "data": { + "id": "sdr-vzub48Y4f7sFBk7J", + "type": "stack-deployment-runs", + } + }, + }, + } + + # ── Model tests ────────────────────────────────────────────────────────── + + def test_stack_deployment_parse(self, stack_deployment_api_data): + """StackDeployment parses id, name, and the stack relation.""" + attrs = dict(stack_deployment_api_data["attributes"]) + attrs["id"] = stack_deployment_api_data["id"] + deployment = StackDeployment.model_validate(attrs) + assert deployment.id == "st-MWvJsvy1FCg3bnXY-std-simple" + assert deployment.name == "simple" + + def test_list_options_serialization(self): + """StackDeploymentListOptions serializes page[size] and include.""" + opts = StackDeploymentListOptions( + page_size=50, + include=[StackDeploymentIncludeOpt.LATEST_DEPLOYMENT_RUN], + ) + dumped = opts.model_dump(by_alias=True, exclude_none=True) + assert dumped["page[size]"] == 50 + + # ── list() tests ───────────────────────────────────────────────────────── + + def test_list_invalid_stack_id_raises(self, service): + """list() with an empty stack id raises InvalidStackIDError on iteration.""" + with pytest.raises(InvalidStackIDError): + list(service.list("")) + + def test_list_success(self, service, stack_deployment_api_data): + """list() yields StackDeployment objects from paginated results.""" + service._list = Mock(return_value=[stack_deployment_api_data]) + + opts = StackDeploymentListOptions(page_size=20) + results = list(service.list(stack_id="st-MWvJsvy1FCg3bnXY", options=opts)) + + service._list.assert_called_once_with( + path="/api/v2/stacks/st-MWvJsvy1FCg3bnXY/stack-deployments", + params={"page[size]": 20}, + ) + assert len(results) == 1 + assert isinstance(results[0], StackDeployment) + assert results[0].id == "st-MWvJsvy1FCg3bnXY-std-simple" + assert results[0].name == "simple" + + def test_list_hydrates_stack_relation(self, service, stack_deployment_api_data): + """list() parses the stack relationship into a typed Stack stub.""" + service._list = Mock(return_value=[stack_deployment_api_data]) + + results = list(service.list(stack_id="st-MWvJsvy1FCg3bnXY")) + + assert isinstance(results[0].stack, Stack) + assert results[0].stack.id == "st-MWvJsvy1FCg3bnXY" + + def test_list_latest_run_reachable_raw(self, service, stack_deployment_api_data): + """The unmodelled latest-deployment-run relation is reachable losslessly.""" + service._list = Mock(return_value=[stack_deployment_api_data]) + + results = list(service.list(stack_id="st-MWvJsvy1FCg3bnXY")) + + refs = results[0].related("latest-deployment-run") + assert refs[0]["id"] == "sdr-vzub48Y4f7sFBk7J" + # raw escape-hatch data never leaks into model_dump() + assert "relationships" not in results[0].model_dump() + + def test_list_with_include(self, service, stack_deployment_api_data): + """list() passes include param as a comma-separated string.""" + service._list = Mock(return_value=[stack_deployment_api_data]) + + opts = StackDeploymentListOptions( + include=[ + StackDeploymentIncludeOpt.LATEST_DEPLOYMENT_RUN, + StackDeploymentIncludeOpt.LATEST_DEPLOYMENT_RUN_STACK_CONFIGURATION, + ] + ) + list(service.list(stack_id="st-MWvJsvy1FCg3bnXY", options=opts)) + + _, kwargs = service._list.call_args + assert ( + kwargs["params"]["include"] + == "latest_deployment_run,latest_deployment_run.stack_configuration" + ) + + def test_list_empty(self, service): + """list() returns an empty iterator when no items are returned.""" + service._list = Mock(return_value=[]) + + results = list(service.list(stack_id="st-MWvJsvy1FCg3bnXY")) + assert results == [] + + def test_list_no_options(self, service, stack_deployment_api_data): + """list() works correctly when no options are given.""" + service._list = Mock(return_value=[stack_deployment_api_data]) + + results = list(service.list(stack_id="st-MWvJsvy1FCg3bnXY")) + + service._list.assert_called_once_with( + path="/api/v2/stacks/st-MWvJsvy1FCg3bnXY/stack-deployments", + params={}, + ) + assert len(results) == 1