Writing Skills
writing-skills
效率与协作技能,来自 obra/superpowers,查看原文获取完整用途和操作说明。
terraform-sdk【部署与运维】Integrates with Terraform and OpenTofu via the HCP Terraform API (pyTFE), CDKTF Python bindings, and the Terraform Cloud API to manage providers, resources, state, and modules.
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: terraform-sdk description: Integrates with Terraform and OpenTofu via the HCP Terraform API (pyTFE), CDKTF Python bindings, and the Terraform Cloud API to manage providers, resources, state, and modules. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding triggers: terraform api, terraform cloud, cdktf python, terraform provider, opentofu, terraform state, terraform modules, hcp terraform archetypes:
Integrates with Terraform and OpenTofu using the HCP Terraform API (via pyTFE), the CDKTF Python SDK, and the Terraform Cloud/Enterprise REST API to programmatically manage providers, resources, state, workspaces, and modules.
pyTFE (pytfe.TFEClient) for HCP Terraform and Terraform Enterprise API operationscdktf with cdktf get to generate provider bindingsworkspaces and state-versions endpointstfe Terraform provider if you need to manage Terraform Cloud resources in HCLTFE_TOKEN environment variable or explicit TFEConfig(token=...)Pulumi (see coding-pulumi skill) as an alternative when you prefer native IaC over Terraform wrappersUse this skill when:
Avoid this skill for:
boto3 for AWS)coding-docker-api)Authenticate to Terraform Cloud/Enterprise — Create a TFEClient with a TFEConfig containing the address and API token. Checkpoint: Call client.organizations.list() to verify connectivity and permissions.
Create or Select Workspace — Use client.workspaces.create() for new workspaces or client.workspaces.list() to find existing ones. Checkpoint: Verify workspace ID and that the execution mode matches your intent (local vs. remote).
Set Variables and Configuration — Use client.variables.create() for terraform variables and environment variables. Checkpoint: Confirm variable values are correct (mark sensitive variables as sensitive).
Trigger a Run — Create a new run via client.runs.create() with a configuration version. Monitor the run through its state machine: pending → planning → applying → applied. Checkpoint: Poll run.status and handle planning errors or apply failures.
Retrieve Outputs and State — Use client.state_versions.current() to get the current state and client.run.outputs() to extract terraform output values. Checkpoint: Validate outputs match expected schema.
from constructs import Construct
from cdktf import App, TerraformStack, TerraformOutput
from cdktf_cdktf_provider_aws.s3_bucket import S3Bucket
from cdktf_cdktf_provider_aws.provider import AwsProvider
class S3BucketStack(TerraformStack):
"""Define an S3 bucket using CDKTF Python bindings."""
def __init__(self, scope: Construct, id: str, bucket_name: str):
super().__init__(scope, id)
AwsProvider(self, "aws", region="us-east-1")
bucket = S3Bucket(
self,
"my-bucket",
bucket=bucket_name,
versioning={
"enabled": True,
},
tags={
"Name": bucket_name,
"ManagedBy": "CDKTF",
},
)
TerraformOutput(self, "bucket_arn", value=bucket.arn)
TerraformOutput(self, "bucket_id", value=bucket.id)
def deploy_s3_bucket(bucket_name: str) -> None:
"""Deploy an S3 bucket programmatically using CDKTF.
Requires: pip install cdktf cdktf-cdktf-provider-aws
Run with: python app.py && cdktf deploy
"""
app = App()
S3BucketStack(app, "s3-bucket-stack", bucket_name)
app.synth()
import os
import time
from pytfe import TFEClient, TFEConfig
from pytfe.exceptions import TFEException
def create_workspace_and_run(
organization: str,
workspace_name: str,
working_directory: str = "",
variables: dict | None = None,
auto_apply: bool = True,
wait_for_completion: bool = True,
) -> dict:
"""Create a Terraform Cloud workspace and trigger a run.
Args:
organization: HCP Terraform organization name.
workspace_name: Name for the new workspace.
working_directory: Relative path to terraform config.
variables: Terraform variables to set.
auto_apply: Whether to auto-apply on plan success.
wait_for_completion: Wait for the run to finish.
Returns:
Dict with workspace ID, run ID, and run status.
Raises:
TFEException: If API operations fail.
TimeoutError: If the run doesn't complete.
"""
config = TFEConfig(
address=os.environ.get("TFE_ADDRESS", "https://app.terraform.io"),
token=os.environ["TFE_TOKEN"],
)
client = TFEClient(config)
# Create or find the workspace
try:
workspace = client.workspaces.create(
organization=organization,
name=workspace_name,
auto_apply=auto_apply,
working_directory=working_directory or None,
)
except TFEException as exc:
# Workspace might already exist
if "already exists" in str(exc):
for ws in client.workspaces.list(organization).items:
if ws.name == workspace_name:
workspace = ws
break
else:
raise
# Set variables
for key, value in (variables or {}).items():
client.variables.create(
workspace_id=workspace.id,
key=key,
value=str(value),
category="terraform",
)
# Create a new configuration version (upload source)
cv = client.configuration_versions.create(
workspace_id=workspace.id,
auto_queue_runs=True,
)
# Trigger the run
run = client.runs.create(workspace_id=workspace.id)
if wait_for_completion:
_wait_for_run(client, run.id)
return {
"workspace_id": workspace.id,
"workspace_name": workspace.name,
"run_id": run.id,
"status": run.status,
}
def _wait_for_run(client: TFEClient, run_id: str, timeout: int = 600) -> None:
"""Poll a Terraform Cloud run until it completes or errors."""
terminal_states = {"applied", "planned_and_finished", "errored", "canceled"}
start = time.monotonic()
while time.monotonic() - start < timeout:
run = client.runs.read(run_id)
if run.status in terminal_states:
if run.status == "errored":
raise RuntimeError(
f"Terraform run {run_id} failed with status: {run.status}"
)
return
if run.status == "planning":
_check_run_errors(run)
time.sleep(5)
raise TimeoutError(f"Terraform run {run_id} did not complete in {timeout}s.")
import os
from pytfe import TFEClient, TFEConfig
from pytfe.exceptions import TFEException
def get_workspace_outputs(
organization: str,
workspace_name: str,
) -> dict:
"""Retrieve Terraform outputs from a Cloud workspace's current state.
Args:
organization: HCP Terraform organization.
workspace_name: Workspace name.
Returns:
Dict of output names to their values.
Raises:
RuntimeError: If workspace or state is not found.
"""
config = TFEConfig(
address=os.environ.get("TFE_ADDRESS", "https://app.terraform.io"),
token=os.environ["TFE_TOKEN"],
)
client = TFEClient(config)
# Find the workspace
workspace = None
for ws in client.workspaces.list(organization).items:
if ws.name == workspace_name:
workspace = ws
break
if workspace is None:
raise RuntimeError(
f"Workspace '{workspace_name}' not found in organization '{organization}'."
)
# Get current state version
try:
state_version = client.state_versions.current(workspace_id=workspace.id)
except TFEException as exc:
raise RuntimeError(
f"Failed to retrieve current state for '{workspace_name}': {exc}"
) from exc
# Parse and return outputs
outputs: dict = {}
if state_version and state_version.outputs:
for output_name, output_data in state_version.outputs.items():
outputs[output_name] = output_data.get("value")
return outputs
# ❌ BAD — silent pass, no state awareness
def get_workspace_bad(client, org):
try:
ws = client.workspaces.create(organization=org, name="my-ws")
return ws
except Exception:
pass # Ignores "already exists" and other failures
# ✅ GOOD — idempotent create-or-find with explicit error handling
def get_or_create_workspace(
client: TFEClient,
organization: str,
name: str,
) -> dict:
"""Retrieve or create a Terraform Cloud workspace.
Args:
client: Authenticated TFEClient instance.
organization: Organization name.
name: Workspace name.
Returns:
Workspace object.
Raises:
RuntimeError: If creation fails for reasons other than conflict.
"""
try:
workspace = client.workspaces.create(organization=organization, name=name)
return {"id": workspace.id, "name": workspace.name, "created": True}
except TFEException as exc:
if "already exists" in str(exc):
for ws in client.workspaces.list(organization).items:
if ws.name == name:
return {"id": ws.id, "name": ws.name, "created": False}
raise RuntimeError(
f"Failed to create workspace '{name}': {exc}"
) from exc
TFE_TOKEN, TFE_ADDRESS) for authentication — never hardcode tokenscategory="terraform" for Terraform variables and category="env" for environment variablesauto_apply=True for workspaces where manual approval is not needed in CI/CDsensitive=true (e.g., database passwords, API keys)| Skill | Purpose |
|---|---|
coding-pulumi | Pulumi Automation API for Python-native IaC |
coding-kubernetes-api | Kubernetes provider via Terraform or client-python |
coding-ansible-api | Ansible for configuration management with provisioned infrastructure |