python-utilities-tfc

tf_utils.py

A Python utility module for advanced management and automation of Terraform Cloud workspaces and state files.

Features

• State File Filtering:
  Clean and filter Terraform state files by resource/module prefixes.

• Terraform Cloud API Automation:

  • Download/upload state files.
  • Lock/unlock workspaces.
  • Trigger, monitor, and manage runs (cancel, discard, apply).
  • Manage workspace variables.
  • Retrieve run outputs and plan summaries.

• Logging:
  Consistent, timestamped logging for all operations.

• SSL Verification: SSL verification is skipped if there is any certificate errors.

Usage

1. Setup

Install dependencies:

pip3 install requests
pip3 install python-utilities-tfc

Import the module in your Python script:

from python_utilities_tfc import tf_utils

export TERRAFORM_CLOUD_TOKEN="your-token"
export TERRAFORM_ORG_NAME="your-org-name"

2. Example Operations

Projects Details

import os
import logging

from tf_utils import (
    setup_logging,
    get_all_projects,
    get_project_by_id,
    log_workspace_project_mapping,
    get_workspaces_by_project
)

def main():
    setup_logging()

    projects = get_all_projects()
    if not projects:
        logging.error("No projects found.")
        return

    for project in projects:
        project_id = project.get("id")
        project_name = project.get("attributes", {}).get("name", "N/A")
        workspaces = get_workspaces_by_project(project_id)

        if not workspaces:
            logging.info(f"No workspaces found for project '{project_name}' ({project_id})")
            continue

        log_workspace_project_mapping(workspaces)

if __name__ == "__main__":
    main()

Auto-Apply-Workspace

import os
import logging

from tf_utils import (
    setup_logging,
    trigger_run,
    monitor_run,
    get_plan_summary,
    apply_run,
    get_workspace_id
)

def main():
    setup_logging()

    workspace_name = os.getenv("TARGET_WORKSPACE_NAME", "Virtual-Machines-Management")

    # Read safe prefixes from environment or fallback to "null_resource"
    safe_prefixes = os.getenv("SAFE_PREFIXES", "null_resource").split(",")

    try:
        # Step 1: Get Workspace ID
        workspace_id = get_workspace_id(workspace_name)

        # Step 2: Trigger Plan-only Run
        logging.info(f"🚀 Triggering a plan-only run on workspace: {workspace_name}")
        run_id = trigger_run(workspace_id, auto_apply=False)

        if not run_id:
            logging.error("❌ Failed to trigger run. No run ID returned.")
            return

        # Step 3: Monitor the Plan
        logging.info(f"🔎 Monitoring plan run {run_id}...")
        monitor_run(run_id)

        # Step 4: Summarize the Plan
        plan_result = get_plan_summary(run_id)

        if not plan_result:
            logging.warning("⚠️ No plan summary available. Cannot decide to auto-apply.")
            return

        summary = plan_result.get("summary", {})
        resource_changes = plan_result.get("resource_changes", [])

        adds = summary.get("add", 0)
        changes = summary.get("change", 0)
        destroys = summary.get("destroy", 0)
        has_changes = summary.get("has-changes", True)

        # Step 5: Log resource changes
        if resource_changes:
            logging.info("🛠️ The following resources will be created/updated/deleted:")
            for res in resource_changes:
                logging.info(f"- {res['address']} ({', '.join(res['actions'])})")
        else:
            logging.info("✅ No actionable resource changes (only 'no-op').")

        # Step 6: Evaluate changes
        unsafe_changes = []
        for res in resource_changes:
            actions = res.get("actions", res.get("change", {}).get("actions", []))
            address = res["address"]

            for action in actions:
                if action == "create":
                    continue  # always allowed
                elif action in ["delete", "update"]:
                    if not any(address.startswith(prefix) for prefix in safe_prefixes):
                        unsafe_changes.append({
                            "address": address,
                            "actions": actions
                        })
                else:
                    # Any unknown action -> treat as unsafe
                    unsafe_changes.append({
                        "address": address,
                        "actions": actions
                    })

        # Step 7: Auto-Apply if Safe
        logging.info(f"🛠 Plan Summary: {adds} to add, {changes} to change, {destroys} to destroy.")

        if has_changes:
            if not unsafe_changes:
                logging.info("✅ Only safe changes detected. Proceeding to auto-apply...")
                apply_run(run_id)

                # Step 8: Monitor the Apply
                logging.info(f"🔎 Monitoring apply for run {run_id}...")
                monitor_run(run_id)
                logging.info("✅ Apply completed successfully.")
            else:
                logging.warning("⚠️ Unsafe resource changes detected. Manual review required:")
                for unsafe in unsafe_changes:
                    logging.warning(f"- {unsafe['address']} ({', '.join(unsafe['actions'])})")
                logging.warning("🚫 Skipping auto-apply due to unsafe changes.")
        else:
            logging.info("✅ No changes detected. No need to apply.")

    except Exception as e:
        logging.error(f"❌ Error occurred: {e}")

if __name__ == "__main__":
    main()

Clean a State File

tf_utils.clean_state_file_by_prefixes(
    input_path="input.tfstate",
    output_path="cleaned.tfstate",
    keep_prefixes=["module1"],
    remove_prefixes=["module2"]
)

Download Workspace State

tf_utils.download_workspace_state(
    workspace_name="my-workspace",
)

Trigger and Monitor a Run

run_id = tf_utils.trigger_run(
    workspace_id="ws-xxxx",
    auto_apply=True
)
tf_utils.monitor_run(
    run_id=run_id
)

Get Run Outputs

outputs = tf_utils.get_run_outputs(
    run_id="run-xxxx"
)
print(outputs)

Get Workspace Runs

import os
import logging

from tf_utils import (
    setup_logging,
    get_workspace_id,
    get_workspace_runs
)

def main():
    setup_logging()

    workspace_name = os.getenv("TARGET_WORKSPACE_NAME", "Virtual-Machines-Management")

    try:
        workspace_id = get_workspace_id(workspace_name)
        if not workspace_id:
            logging.error(f"Workspace '{workspace_name}' not found.")
            return
        logging.info(f"Workspace ID for '{workspace_name}': {workspace_id}")
        runs = get_workspace_runs(workspace_id)
        if not runs:
            logging.info(f"No runs found for workspace '{workspace_name}' ({workspace_id})")
            return
        logging.info(f"Total Runs: {len(runs)}")

    except Exception as e:
        logging.error(f"An error occurred: {e}")

if __name__ == "__main__":
    main()

Functions Overview

• setup_logging(): Configure logging.
• clean_state_file_by_prefixes(): Filter resources in a state file.
• download_workspace_state(): Download the latest state from Terraform Cloud.
• push_state_to_terraform_cloud(): Upload a state file to Terraform Cloud.
• lock_workspace(), unlock_workspace(): Lock/unlock a workspace.
• trigger_run(), monitor_run(), cancel_run(), discard_run(), apply_run(): Manage runs.
• get_run_outputs(), get_plan_summary(): Retrieve outputs and plan details.
• get_variable_id(), update_variable(): Manage workspace variables.

Requirements

• Python 3.11+
• requests library

Notes

• All API operations require a valid Terraform Cloud API token.
• Use caution when discarding or canceling runs, as this may affect your infrastructure state.

License

MIT License