### Install TofuPy Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Installs the tofupy library using pip. Requires either tofu or terraform to be installed and available in the system's PATH. ```bash pip install tofupy ``` -------------------------------- ### TofuPy Quick Start: Initialize, Validate, Plan, Apply, Output Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Demonstrates the basic workflow of using the TofuPy library to manage infrastructure. It covers initializing a workspace, validating the configuration, creating a plan, applying the changes, and retrieving outputs. ```python from tofupy import Tofu # Initialize a workspace workspace = Tofu(cwd="/path/to/terraform/config") # Initialize Terraform workspace.init() # Validate configuration validation = workspace.validate() if not validation.valid: print("Configuration is invalid!") for diagnostic in validation.diagnostics: print(f"Error: {diagnostic.summary}") # Create and review a plan plan_log, plan = workspace.plan() if plan and not plan.errored: print(f"Plan will create {len([c for c in plan.resource_changes.values() if 'create' in c.change.actions])} resources") # Apply changes apply_log = workspace.apply() print(f"Applied: {apply_log.added} added, {apply_log.changed} changed, {apply_log.removed} removed") # Get outputs outputs = workspace.output() for name, output in outputs.items(): print(f"{name}: {output.value} (type: {output.type})") ``` -------------------------------- ### Manage Terraform Environments with Tofu Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Automates the setup and deployment of Terraform infrastructure across multiple environments. It leverages the Tofu library to initialize workspaces, validate configurations, plan changes, and apply deployments. Supports environment-specific backend configurations and variable injection. ```Python from tofupy import Tofu from pathlib import Path class InfrastructureManager: def __init__(self, base_path): self.base_path = Path(base_path) self.environments = {} def setup_environment(self, env_name, variables=None): """Initialize a Terraform workspace for an environment""" env_path = self.base_path / env_name workspace = Tofu( cwd=str(env_path), env={f"TF_VAR_{k}": str(v) for k, v in (variables or {}).items()} ) # Initialize with environment-specific backend backend_config = env_path / "backend.hcl" if backend_config.exists(): workspace.init(backend_conf=backend_config) else: workspace.init() self.environments[env_name] = workspace return workspace def deploy_all(self, environments=None): """Deploy to multiple environments""" target_envs = environments or self.environments.keys() for env_name in target_envs: print(f"\nšŸš€ Deploying to {env_name}...") workspace = self.environments[env_name] # Validate first validation = workspace.validate() if not validation.valid: print(f"āŒ {env_name} configuration invalid") continue # Plan changes plan_log, plan = workspace.plan() if not plan or plan.errored: print(f"āŒ {env_name} planning failed") continue # Show summary creates = len([c for c in plan.resource_changes.values() if 'create' in c.change.actions]) updates = len([c for c in plan.resource_changes.values() if 'update' in c.change.actions]) deletes = len([c for c in plan.resource_changes.values() if 'delete' in c.change.actions]) print(f" Plan: +{creates} ~{updates} -{deletes}") # Apply changes apply_log = workspace.apply() if apply_log.errors: print(f"āŒ {env_name} apply failed") for error in apply_log.errors: print(f" {error.summary}") else: print(f"āœ… {env_name} deployed successfully") # Usage manager = InfrastructureManager("./environments") # Setup environments manager.setup_environment("dev", {"instance_type": "t3.micro"}) manager.setup_environment("staging", {"instance_type": "t3.small"}) manager.setup_environment("prod", {"instance_type": "t3.medium"}) # Deploy to all environments manager.deploy_all() ``` -------------------------------- ### Initialize Terraform Workspace Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Demonstrates initializing a Terraform workspace using TofuPy. Covers basic initialization, initialization with backend configuration, initialization without backends, and initialization with extra arguments. ```Python from tofupy import Tofu workspace = Tofu(cwd="./terraform") # Basic initialization success = workspace.init() # Initialize with backend configuration success = workspace.init(backend_conf="backend.hcl") # Initialize without backend (useful for validation) success = workspace.init(disable_backends=True) # Initialize with extra arguments success = workspace.init(extra_args=["-upgrade"]) ``` -------------------------------- ### TofuPy Core: Tofu Class Initialization and Configuration Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Shows how to instantiate the Tofu class for interacting with Terraform/OpenTofu. Covers basic initialization using the current directory and advanced configuration options like specifying the working directory, binary path, log level, and environment variables. ```python from tofupy import Tofu # Basic initialization workspace = Tofu() # Uses current directory # Custom configuration workspace = Tofu( cwd="/path/to/terraform/config", binary="terraform", # or "tofu" log_level="DEBUG", # ERROR, WARN, INFO, DEBUG env={"TF_VAR_environment": "production"} ) # The workspace automatically detects binary version and validates compatibility print(f"Using {workspace.binary_path} version {workspace.version} on {workspace.platform}") ``` -------------------------------- ### Plan Terraform Changes Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Illustrates how to create a Terraform execution plan with TofuPy. Covers basic planning, planning with variables, planning with output to a file, and planning with extra arguments. Also shows how to analyze the plan details. ```Python # Basic plan plan_log, plan = workspace.plan() # Plan with variables plan_log, plan = workspace.plan( variables={ "environment": "production", "instance_count": "3" } ) # Plan with output to file from pathlib import Path plan_file = Path("./my-plan.tfplan") plan_log, plan = workspace.plan(plan_file=plan_file) # Plan with extra arguments plan_log, plan = workspace.plan( extra_args=["-target=aws_instance.example"] ) # Analyze the plan if plan and not plan.errored: print(f"Terraform version: {plan.terraform_version}") print(f"Plan is applyable: {plan.applyable}") # Review resource changes for address, change_container in plan.resource_changes.items(): change = change_container.change actions = ", ".join(change.actions) print(f"{address}: {actions}") # Check for creates if "create" in change.actions: print(f" Will create {change_container.type} resource") # Check for updates if "update" in change.actions: print(f" Will modify {change_container.type} resource") # Check for destroys if "delete" in change.actions: print(f" Will destroy {change_container.type} resource") ``` -------------------------------- ### TofuPy API Reference: Tofu Class Constructor Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Details the constructor for the Tofu class, outlining parameters for specifying the working directory, binary path, log level, and environment variables. ```APIDOC Tofu(cwd=None, binary="tofu", log_level="ERROR", env={}) ``` -------------------------------- ### TofuPy Configuration: Binary Selection Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Specifies which binary TofuPy should use, allowing selection of OpenTofu ('tofu'), a specific Terraform executable path, or automatic detection. ```python # Use OpenTofu workspace = Tofu(binary="tofu") # Use specific Terraform path workspace = Tofu(binary="/usr/local/bin/terraform") # Let TofuPy find the binary workspace = Tofu() # Tries "tofu" first, then "terraform" ``` -------------------------------- ### Apply Terraform Changes Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Demonstrates applying Terraform changes using TofuPy. Covers applying from a plan file, direct apply with variables, and applying with auto-approval. Includes reviewing apply results and handling errors. ```Python # Apply from plan file apply_log = workspace.apply(plan_file=plan_file) # Direct apply with variables apply_log = workspace.apply( variables={"environment": "staging"} ) # Apply with auto-approval (default) apply_log = workspace.apply() # Review apply results print(f"Operation: {apply_log.operation}") print(f"Resources: +{apply_log.added} ~{apply_log.changed} -{apply_log.removed}") # Handle errors if apply_log.errors: print("Apply encountered errors:") for error in apply_log.errors: print(f" {error.summary}: {error.detail}") ``` -------------------------------- ### TofuPy API Reference: Tofu Class Methods Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Lists the core methods available on the Tofu class for managing Terraform/OpenTofu operations, including initialization, validation, planning, applying, destroying, state management, and output retrieval. ```APIDOC init(disable_backends=False, backend_conf=None, extra_args=[]) → bool validate() → Validate plan(variables={}, plan_file=None, event_handlers={}, extra_args=[]) → Tuple[PlanLog, Plan | None] apply(plan_file=None, variables={}, destroy=False, event_handlers={}, extra_args=[]) → ApplyLog destroy() → ApplyLog state() → State output() → Dict[str, Output] ``` -------------------------------- ### Manage Terraform State Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Explains how to interact with Terraform state using TofuPy. Shows how to retrieve the current state, access state outputs, and iterate through resources within the root module. ```Python # Get current state state = workspace.state() print(f"State serial: {state.serial}") print(f"State lineage: {state.lineage}") print(f"Terraform version: {state.terraform_version}") # Access state outputs for name, output in state.outputs.items(): if output.sensitive: print(f"{name}: ") else: print(f"{name}: {output.value} ({output.type})") # Access state resources root_module = state.root_module if root_module: for address, resource in root_module.resources.items(): print(f"Resource: {address}") print(f" Type: {resource.type}") print(f" Provider: {resource.provider_name}") print(f" Values: {resource.values}") ``` -------------------------------- ### Retrieve Terraform Outputs Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Demonstrates how to fetch Terraform outputs using TofuPy. Covers retrieving all outputs and accessing specific outputs by name, including handling sensitive output values. ```Python # Get all outputs outputs = workspace.output() for name, output in outputs.items(): print(f"Output '{name}':") print(f" Value: {output.value}") print(f" Type: {output.type}") print(f" Sensitive: {output.sensitive}") # Work with specific outputs if "database_url" in outputs: db_url = outputs["database_url"].value print(f"Connecting to database: {db_url}") ``` -------------------------------- ### TofuPy API Reference: Data Models Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Describes the key data models provided by TofuPy for structured access to Terraform's JSON outputs, including detailed information on Plan, State, Logs, and Resource changes. ```APIDOC Plan: Resource changes, drift detection, variables, and metadata State: Current infrastructure state with resources and outputs ApplyLog/PlanLog: Execution logs with summaries and diagnostics Validate: Configuration validation results Output: Terraform output values with type information Resource: Individual infrastructure resources Change: Detailed before/after states for resource modifications ``` -------------------------------- ### Analyze Terraform State with Tofu Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Generates a comprehensive infrastructure report by analyzing Terraform state files. It extracts information about the Terraform version, state lineage, resource distribution by provider and type, and output values. This helps in understanding the current infrastructure composition. ```Python def generate_infrastructure_report(workspace_path): """Generate a comprehensive infrastructure report""" workspace = Tofu(cwd=workspace_path) # Get current state state = workspace.state() print("=== Infrastructure Report ===\n") print(f"Terraform Version: {state.terraform_version}") print(f"State Serial: {state.serial}") print(f"State Lineage: {state.lineage}") # Analyze resources by provider providers = {} resource_types = {} if state.root_module: for address, resource in state.root_module.resources.items(): provider = resource.provider_name or "unknown" providers[provider] = providers.get(provider, 0) + 1 resource_types[resource.type] = resource_types.get(resource.type, 0) + 1 print(f"\n=== Resources by Provider ===") for provider, count in sorted(providers.items()): print(f" {provider}: {count} resources") print(f"\n=== Resources by Type ===") for rtype, count in sorted(resource_types.items()): print(f" {rtype}: {count}") # List outputs print(f"\n=== Outputs ===") if state.outputs: for name, output in state.outputs.items(): status = "sensitive" if output.sensitive else "public" print(f" {name} ({output.type}): {status}") else: print(" No outputs defined") # Usage generate_infrastructure_report("./terraform") ``` -------------------------------- ### Validate Terraform Configuration Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Shows how to validate Terraform configuration files using TofuPy. Includes checking the validation status, error counts, and iterating through diagnostics to report errors and warnings. ```Python # Validate configuration validation = workspace.validate() print(f"Configuration valid: {validation.valid}") print(f"Errors: {validation.error_count}, Warnings: {validation.warning_count}") # Handle validation issues if not validation.valid: for diagnostic in validation.diagnostics: if diagnostic.severity == "error": print(f"āŒ {diagnostic.summary}") print(f" {diagnostic.detail}") else: print(f"āš ļø {diagnostic.summary}") ``` -------------------------------- ### Destroy Terraform Infrastructure Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Shows how to destroy all managed infrastructure using TofuPy. Includes executing the destroy command and handling any errors that occur during the process. ```Python # Destroy all resources destroy_log = workspace.destroy() print(f"Destroyed {destroy_log.removed} resources") # Handle destruction errors if destroy_log.errors: print("Destroy encountered errors:") for error in destroy_log.errors: print(f" {error.summary}") ``` -------------------------------- ### TofuPy Event Handlers for Terraform Operations Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Demonstrates how to use TofuPy's event handling mechanism to monitor Terraform's plan and apply operations in real-time. It includes handlers for progress, errors, and change summaries, allowing for custom logging and feedback during execution. ```Python def progress_handler(event): """Handle progress events during plan/apply""" if event.get("type") == "apply_progress": resource = event.get("hook", {}).get("resource", {}) action = resource.get("action") addr = resource.get("addr") print(f"Progress: {action} {addr}") return True def error_handler(event): """Handle error events""" if event.get("@level") == "error": print(f"Error: {event.get('@message')}") return True def summary_handler(event): """Handle change summaries""" if event.get("type") == "change_summary": changes = event.get("changes", {}) add = changes.get("add", 0) change = changes.get("change", 0) remove = changes.get("remove", 0) print(f"Summary: +{add} ~{change} -{remove}") return True # Use event handlers during operations plan_log, plan = workspace.plan( event_handlers={ "apply_progress": progress_handler, "diagnostic": error_handler, "change_summary": summary_handler, "all": lambda e: print(f"Event: {e.get('type')}") # Catch all events } ) apply_log = workspace.apply( event_handlers={ "apply_progress": progress_handler, "change_summary": summary_handler } ) ``` -------------------------------- ### Tofu/Terraform Machine Readable UI Source: https://github.com/terraformindepth/tofupy/blob/main/docs/dev/tofu_ui.md Describes the machine-readable UI format used for streaming events in OpenTofu and Terraform. Each event is a complete JSON object on a new line. ```APIDOC Machine Readable UI: - Format: JSON object per line - Usage: Streaming events (plan, apply) - Reference: https://developer.hashicorp.com/terraform/internals/machine-readable-ui ``` -------------------------------- ### TofuPy Configuration: Logging and Debugging Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Sets the logging level for TofuPy operations (e.g., DEBUG, INFO, WARN, ERROR) and notes that TF_IN_AUTOMATION is automatically set. ```python # Set Terraform log level workspace = Tofu(log_level="DEBUG") # ERROR, WARN, INFO, DEBUG # TofuPy automatically sets TF_IN_AUTOMATION=1 for consistent output ``` -------------------------------- ### TofuPy Error Handling Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Demonstrates how to handle validation and apply errors using TofuPy's diagnostic system. Catches potential exceptions like FileNotFoundError and RuntimeError for robust operation. ```python from tofupy import Tofu, Diagnostic workspace = Tofu(cwd="./terraform") try: workspace.init() # Validation errors validation = workspace.validate() if not validation.valid: print("Configuration errors:") for diagnostic in validation.diagnostics: if diagnostic.severity == "error": print(f" āŒ {diagnostic.summary}") print(f" {diagnostic.detail}") # Apply errors apply_log = workspace.apply() if apply_log.errors: print("Apply errors:") for error in apply_log.errors: print(f" āŒ {error.summary}") if error.detail: print(f" {error.detail}") except FileNotFoundError as e: print(f"Terraform/OpenTofu binary not found: {e}") except RuntimeError as e: print(f"Terraform operation failed: {e}") ``` -------------------------------- ### TofuPy Configuration: Environment Variables Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Configures the TofuPy workspace with custom environment variables, including Terraform variables (TF_VAR_), AWS profiles, and logging levels (TF_LOG). ```python workspace = Tofu( env={ "TF_VAR_region": "us-west-2", "TF_VAR_environment": "production", "AWS_PROFILE": "production", "TF_LOG": "DEBUG" # Override log level } ) ``` -------------------------------- ### Tofu/Terraform JSON Output Format Source: https://github.com/terraformindepth/tofupy/blob/main/docs/dev/tofu_ui.md Details the JSON output format used for non-streaming events, such as the output of the 'tofu show' command. ```APIDOC JSON Output Format: - Usage: Non-streaming events (e.g., 'tofu show') - Reference: https://developer.hashicorp.com/terraform/internals/json-format ``` -------------------------------- ### TofuPy for Infrastructure Security Scanning Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Provides a Python script using TofuPy to scan Terraform infrastructure for security vulnerabilities, specifically focusing on overly permissive security group rules. It analyzes the plan output to identify rules allowing all traffic from the internet or specific ports like SSH. ```Python from tofupy import Tofu def scan_security_groups(workspace_path): """Scan for overly permissive security groups""" workspace = Tofu(cwd=workspace_path) workspace.init() plan_log, plan = workspace.plan() if not plan or plan.errored: print("āŒ Planning failed") return issues = [] for address, change_container in plan.resource_changes.items(): if change_container.type == "aws_security_group_rule": change = change_container.change # Check for rules allowing all traffic from internet if change.after and change.after.get("cidr_blocks") == ["0.0.0.0/0"]: if change.after.get("from_port") == 0 and change.after.get("to_port") == 65535: issues.append(f"āš ļø {address} allows all traffic from internet") elif change.after.get("from_port") == 22: issues.append(f"āš ļø {address} allows SSH from internet") if issues: print("Security issues found:") for issue in issues: print(f" {issue}") else: print("āœ… No security issues detected") # Usage scan_security_groups("./infrastructure") ``` -------------------------------- ### Custom Infrastructure Policy Validation Source: https://github.com/terraformindepth/tofupy/blob/main/README.md Validates planned infrastructure against custom policies, such as ensuring EC2 instances have Name tags, RDS instances are not publicly accessible, and S3 buckets have versioning enabled. Returns True if all policies pass, False otherwise. ```python def validate_infrastructure_policies(workspace_path): """Run custom policy validations on planned infrastructure""" workspace = Tofu(cwd=workspace_path) workspace.init() plan_log, plan = workspace.plan() if not plan or plan.errored: print("āŒ Cannot validate - planning failed") return False violations = [] # Policy: All EC2 instances must have Name tags for address, change_container in plan.resource_changes.items(): if change_container.type == "aws_instance": change = change_container.change if change.after: tags = change.after.get("tags", {}) if "Name" not in tags: violations.append(f"EC2 instance {address} missing Name tag") # Policy: No publicly accessible RDS instances for address, change_container in plan.resource_changes.items(): if change_container.type == "aws_db_instance": change = change_container.change if change.after and change.after.get("publicly_accessible"): violations.append(f"RDS instance {address} is publicly accessible") # Policy: S3 buckets must have versioning enabled for address, change_container in plan.resource_changes.items(): if change_container.type == "aws_s3_bucket_versioning": change = change_container.change if change.after: config = change.after.get("versioning_configuration", [{}])[0] if config.get("status") != "Enabled": violations.append(f"S3 bucket versioning not enabled for {address}") # Report results if violations: print("āŒ Policy violations found:") for violation in violations: print(f" • {violation}") return False else: print("āœ… All policy checks passed") return True # Usage if validate_infrastructure_policies("./terraform"): print("Proceeding with deployment...") else: print("Fix policy violations before deploying") ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.