python-utilities-tfc 1.9.0
Utility functions to work with Terraform Cloud and manage Terraform state.
Requires Python: >=3.11

# 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:

```bash
pip3 install requests
pip3 install python-utilities-tfc
```

Import the module in your Python script:

```python
from python_utilities_tfc import tf_utils
```

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

### 2. Example Operations

#### Projects Details

```python

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

```python
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

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

#### Download Workspace State

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

#### Trigger and Monitor a Run

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

#### Get Run Outputs

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

#### Get Workspace Runs

```python
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