Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
224 changes: 224 additions & 0 deletions examples/stack.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,224 @@
# Copyright IBM Corp. 2025, 2026
# SPDX-License-Identifier: MPL-2.0

from __future__ import annotations

import argparse
import os

from pytfe import TFEClient, TFEConfig
from pytfe.models.agent import AgentPool
from pytfe.models.project import Project
from pytfe.models.stack import (
StackCreateOptions,
StackListOptions,
StackSortColumn,
StackUpdateOptions,
StackVcsRepoOptions,
)


def _print_header(title: str):
print("\n" + "=" * 80)
print(title)
print("=" * 80)


def _print_stack(item):
print(f"- id: {item.id}")
print(f"- name: {item.name}")
print(f"- description: {item.description}")
print(f"- created_at: {item.created_at}")
print(f"- updated_at: {item.updated_at}")
print(f"- speculation_enabled: {item.speculation_enabled}")
print(f"- project_id: {item.project.id if item.project else None}")
print(f"- agent_pool_id: {item.agent_pool.id if item.agent_pool else None}")

if item.vcs_repo:
print("- vcs_repo:")
print(f" identifier={item.vcs_repo.identifier}")
print(f" branch={item.vcs_repo.branch}")
print(f" github_app_installation_id={item.vcs_repo.gha_installation_id}")
print(f" oauth_token_id={item.vcs_repo.oauth_token_id}")

Check failure

Code scanning / CodeQL

Clear-text logging of sensitive information High

This expression logs
sensitive data (password)
as clear text.
Comment thread
isivaselvan marked this conversation as resolved.
Dismissed


def _build_vcs_repo_options(args) -> StackVcsRepoOptions | None:
if not args.vcs_identifier:
return None

return StackVcsRepoOptions(
identifier=args.vcs_identifier,
branch=args.vcs_branch,
gha_installation_id=args.vcs_github_app_installation_id,
oauth_token_id=args.vcs_oauth_token_id,
)


def main():
parser = argparse.ArgumentParser(
description="Stacks operations demo for python-tfe"
)
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("--organization", help="Organization name (required for list)")
parser.add_argument(
"--operation",
required=True,
choices=["create", "read", "update", "list", "delete", "force-delete"],
help="Operation to execute",
)

parser.add_argument(
"--stack-id", help="Stack ID (required for read/update/delete/force-delete)"
)

parser.add_argument("--name", help="Stack name (required for create)")
parser.add_argument("--description", help="Stack description")
parser.add_argument(
"--speculation-enabled",
type=lambda v: str(v).lower() in ("1", "true", "yes", "y"),
default=None,
help="Enable speculation (true/false)",
)

parser.add_argument(
"--project-id",
help="Project ID (required for create, optional for list filter)",
)
parser.add_argument(
"--agent-pool-id", help="Agent pool ID (optional for create/update)"
)

parser.add_argument(
"--vcs-identifier",
help="VCS repo identifier (e.g. org/repo), optional for create/update",
)
parser.add_argument("--vcs-branch", help="VCS branch")
parser.add_argument(
"--vcs-github-app-installation-id",
help="GitHub App installation ID for VCS repo",
)
parser.add_argument("--vcs-oauth-token-id", help="OAuth token ID for VCS repo")

parser.add_argument("--page-size", type=int, default=20, help="Page size for list")
parser.add_argument(
"--sort",
choices=[item.value for item in StackSortColumn],
default=None,
help="Sort column for list",
)
parser.add_argument(
"--search-name",
default=None,
help="Search stacks by name",
)

args = parser.parse_args()

cfg = TFEConfig(address=args.address, token=args.token)
client = TFEClient(cfg)

if args.operation == "create":
if not args.name:
parser.error("--name is required for operation=create")
if not args.project_id:
parser.error("--project-id is required for operation=create")

_print_header("Creating stack")
options = StackCreateOptions(
name=args.name,
description=args.description,
speculation_enabled=args.speculation_enabled,
vcs_repo=_build_vcs_repo_options(args),
project=Project(id=args.project_id),
agent_pool=AgentPool(id=args.agent_pool_id) if args.agent_pool_id else None,
)
result = client.stacks.create(options)
print("Created stack")
_print_stack(result)
return

if args.operation == "read":
if not args.stack_id:
parser.error("--stack-id is required for operation=read")

_print_header("Reading stack")
result = client.stacks.read(args.stack_id)
print("Retrieved stack")
_print_stack(result)
return

if args.operation == "update":
if not args.stack_id:
parser.error("--stack-id is required for operation=update")
if not any(
[
args.name,
args.description,
args.speculation_enabled is not None,
args.agent_pool_id,
args.vcs_identifier,
args.vcs_branch,
args.vcs_github_app_installation_id,
args.vcs_oauth_token_id,
args.project_id,
]
):
parser.error("Provide at least one field to update")

_print_header("Updating stack")
options = StackUpdateOptions(
name=args.name,
description=args.description,
speculation_enabled=args.speculation_enabled,
vcs_repo=_build_vcs_repo_options(args),
agent_pool=AgentPool(id=args.agent_pool_id) if args.agent_pool_id else None,
project=Project(id=args.project_id) if args.project_id else None,
)
result = client.stacks.update(args.stack_id, options)
print("Updated stack")
_print_stack(result)
return

if args.operation == "list":
if not args.organization:
parser.error("--organization is required for operation=list")

_print_header("Listing stacks")
list_options = StackListOptions(
page_size=args.page_size,
project_id=args.project_id,
sort=StackSortColumn(args.sort) if args.sort else None,
search_by_name=args.search_name,
)

items = list(client.stacks.list(args.organization, list_options))
print(f"Found {len(items)} stacks")
for item in items:
print("-")
_print_stack(item)
return

if args.operation == "delete":
if not args.stack_id:
parser.error("--stack-id is required for operation=delete")

_print_header("Deleting stack")
client.stacks.delete(args.stack_id)
print(f"Deleted stack: {args.stack_id}")
return

if args.operation == "force-delete":
if not args.stack_id:
parser.error("--stack-id is required for operation=force-delete")

_print_header("Force deleting stack")
client.stacks.force_delete(args.stack_id)
print(f"Force deleted stack: {args.stack_id}")
return


if __name__ == "__main__":
main()
2 changes: 2 additions & 0 deletions src/pytfe/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
from .resources.run_task import RunTasks
from .resources.run_trigger import RunTriggers
from .resources.ssh_keys import SSHKeys
from .resources.stack import Stacks
from .resources.state_version_outputs import StateVersionOutputs
from .resources.state_versions import StateVersions
from .resources.variable import Variables
Expand Down Expand Up @@ -104,6 +105,7 @@ def __init__(self, config: TFEConfig | None = None):

# Reserved Tag Key
self.reserved_tag_key = ReservedTagKeys(self._transport)
self.stacks = Stacks(self._transport)

def close(self) -> None:
try:
Expand Down
120 changes: 120 additions & 0 deletions src/pytfe/models/stack.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
# Copyright IBM Corp. 2025, 2026
# SPDX-License-Identifier: MPL-2.0

from __future__ import annotations

from datetime import datetime
from enum import Enum

from pydantic import BaseModel, ConfigDict, Field, model_validator

from ..errors import ERR_REQUIRED_NAME, ERR_REQUIRED_PROJECT
from .agent import AgentPool
from .project import Project


class StackSortColumn(str, Enum):
"""StackSortColumn represents a string that can be used to sort items when using the List method."""

STACK_SORT_BY_NAME = "name"
STACK_SORT_BY_UPDATED_AT = "updated-at"
STACK_SORT_BY_NAME_DESC = "-name"
STACK_SORT_BY_UPDATED_AT_DESC = "-updated-at"


class StackVcsRepo(BaseModel):
"""StackVCSRepo represents the version control system repository for a stack."""

model_config = ConfigDict(populate_by_name=True, validate_by_name=True)

identifier: str = Field(alias="identifier")
branch: str | None = Field(default=None, alias="branch")
gha_installation_id: str | None = Field(
default=None, alias="github-app-installation-id"
)
oauth_token_id: str | None = Field(default=None, alias="oauth-token-id")


class StackVcsRepoOptions(BaseModel):
"""StackVCSRepoOptions represents the options for the version control system repository for a stack."""

model_config = ConfigDict(populate_by_name=True, validate_by_name=True)

identifier: str = Field(alias="identifier")
branch: str | None = Field(default=None, alias="branch")
gha_installation_id: str | None = Field(
default=None, alias="github-app-installation-id"
)
oauth_token_id: str | None = Field(default=None, alias="oauth-token-id")


class Stack(BaseModel):
"""Stack represents a stack in Terraform Cloud."""

model_config = ConfigDict(populate_by_name=True, validate_by_name=True)

id: str
name: str | None = Field(default=None, alias="name")
description: str | None = Field(default=None, alias="description")
created_at: datetime | None = Field(default=None, alias="created-at")
updated_at: datetime | None = Field(default=None, alias="updated-at")
vcs_repo: StackVcsRepo | None = Field(default=None, alias="vcs-repo")
speculation_enabled: bool | None = Field(default=None, alias="speculation-enabled")
upstream_count: int | None = Field(default=None, alias="upstream-count")
downstream_count: int | None = Field(default=None, alias="downstream-count")
inputs_count: int | None = Field(default=None, alias="inputs-count")
outputs_count: int | None = Field(default=None, alias="outputs-count")
creation_source: str | None = Field(default=None, alias="creation-source")

# Relations
project: Project | None = Field(default=None, alias="project")
agent_pool: AgentPool | None = Field(default=None, alias="agent-pool")
# latest_stack_configuration: dict[str, Any] | None = Field(default=None, alias="latest-stack-configuration")


class StackListOptions(BaseModel):
"""StackListOptions represents the options for listing stacks."""

model_config = ConfigDict(populate_by_name=True, validate_by_name=True)

page_size: int | None = Field(default=None, alias="page[size]")
project_id: str | None = Field(default=None, alias="filter[project][id]")
sort: StackSortColumn | None = Field(default=None, alias="sort")
search_by_name: str | None = Field(default=None, alias="search[name]")


class StackCreateOptions(BaseModel):
"""StackCreateOptions represents the options for creating a stack."""

model_config = ConfigDict(populate_by_name=True, validate_by_name=True)

name: str = Field(alias="name")
migration: bool | None = Field(default=None, alias="migration")
description: str | None = Field(default=None, alias="description")
speculation_enabled: bool | None = Field(default=None, alias="speculation-enabled")
vcs_repo: StackVcsRepoOptions | None = Field(default=None, alias="vcs-repo")
project: Project = Field(alias="project")
agent_pool: AgentPool | None = Field(default=None, alias="agent-pool")

@model_validator(mode="after")
def valid(self) -> StackCreateOptions:
if self.name == "":
raise ValueError(ERR_REQUIRED_NAME)

if self.project and self.project.id == "":
raise ValueError(ERR_REQUIRED_PROJECT)

return self


class StackUpdateOptions(BaseModel):
"""StackUpdateOptions represents the options for updating a stack."""

model_config = ConfigDict(populate_by_name=True, validate_by_name=True)

name: str | None = Field(default=None, alias="name")
description: str | None = Field(default=None, alias="description")
speculation_enabled: bool | None = Field(default=None, alias="speculation-enabled")
vcs_repo: StackVcsRepoOptions | None = Field(default=None, alias="vcs-repo")
project: Project | None = Field(default=None, alias="project")
agent_pool: AgentPool | None = Field(default=None, alias="agent-pool")
Loading
Loading