Python与Terraform集成实践指南

一、使用python-terraform库

1. 基本设置

from python_terraform import *
import os
import logging

class TerraformManager:
    def __init__(self, working_dir: str, variables: dict = None):
        self.working_dir = working_dir
        self.variables = variables or {}
        self.tf = Terraform(
            working_dir=working_dir,
            variables=self.variables
        )
        
    def init(self):
        """初始化Terraform"""
        return_code, stdout, stderr = self.tf.init()
        if return_code != 0:
            raise Exception(f"Terraform init failed: {stderr}")
        return stdout

    def plan(self):
        """执行plan"""
        return_code, stdout, stderr = self.tf.plan(
            no_color=IsFlagged,
            refresh=True,
            capture_output=True
        )
        return stdout

    def apply(self, auto_approve=True):
        """应用配置"""
        return_code, stdout, stderr = self.tf.apply(
            skip_plan=True,
            auto_approve=auto_approve
        )
        if return_code != 0:
            raise Exception(f"Terraform apply failed: {stderr}")
        return stdout

    def destroy(self, auto_approve=True):
        """销毁资源"""
        return_code, stdout, stderr = self.tf.destroy(
            auto_approve=auto_approve
        )
        return stdout

2. 实际使用示例

from terraform_manager import TerraformManager

def deploy_infrastructure():
    # 定义变量
    variables = {
        'region': 'us-west-2',
        'instance_type': 't2.micro',
        'environment': 'dev'
    }

    # 初始化管理器
    tf_manager = TerraformManager(
        working_dir='./terraform',
        variables=variables
    )

    try:
        # 初始化
        print("Initializing Terraform...")
        tf_manager.init()

        # 查看计划
        print("Planning infrastructure changes...")
        plan_output = tf_manager.plan()
        print(plan_output)

        # 应用更改
        print("Applying changes...")
        apply_output = tf_manager.apply()
        print(apply_output)

    except Exception as e:
        print(f"Error: {str(e)}")
        raise

if __name__ == "__main__":
    deploy_infrastructure()

二、高级配置管理

1. 配置管理器

import yaml
from typing import Dict, Optional

class ConfigManager:
    def __init__(self, config_file: str):
        self.config_file = config_file
        self.config = self._load_config()

    def _load_config(self) -> Dict:
        with open(self.config_file, 'r') as f:
            return yaml.safe_load(f)

    def get_environment_config(self, env: str) -> Dict:
        return self.config.get('environments', {}).get(env, {})

    def get_terraform_variables(self, env: str) -> Dict:
        env_config = self.get_environment_config(env)
        return {
            'environment': env,
            **env_config.get('terraform_vars', {})
        }

2. 状态管理

import os
import boto3
from datetime import datetime

class StateManager:
    def __init__(self, bucket: str, key_prefix: str):
        self.s3 = boto3.client('s3')
        self.bucket = bucket
        self.key_prefix = key_prefix

    def backup_state(self, env: str):
        """备份Terraform状态到S3"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        source_key = f"{self.key_prefix}/{env}/terraform.tfstate"
        backup_key = f"{self.key_prefix}/{env}/backups/terraform.tfstate.{timestamp}"
        
        try:
            self.s3.copy_object(
                Bucket=self.bucket,
                CopySource={'Bucket': self.bucket, 'Key': source_key},
                Key=backup_key
            )
            return True
        except Exception as e:
            print(f"Backup failed: {str(e)}")
            return False

三、完整项目示例

1. 项目结构

project/
├── terraform/
│   ├── main.tf
│   ├── variables.tf
│   └── outputs.tf
├── src/
│   ├── __init__.py
│   ├── terraform_manager.py
│   ├── config_manager.py
│   └── state_manager.py
├── config/
│   ├── dev.yaml
│   └── prod.yaml
└── deploy.py

2. Terraform配置

provider "aws" {
  region = var.region
}

resource "aws_instance" "example" {
  ami           = var.ami_id
  instance_type = var.instance_type

  tags = {
    Name        = "${var.environment}-instance"
    Environment = var.environment
  }
}

3. 配置文件

environments:
  dev:
    terraform_vars:
      region: us-west-2
      instance_type: t2.micro
      ami_id: ami-0c55b159cbfafe1f0
    aws:
      profile: development

4. 部署脚本

import argparse
from src.terraform_manager import TerraformManager
from src.config_manager import ConfigManager
from src.state_manager import StateManager

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--env', required=True, choices=['dev', 'prod'])
    parser.add_argument('--action', required=True, 
                       choices=['plan', 'apply', 'destroy'])
    args = parser.parse_args()

    # 初始化配置
    config_manager = ConfigManager(f'config/{args.env}.yaml')
    variables = config_manager.get_terraform_variables(args.env)

    # 初始化状态管理
    state_manager = StateManager(
        bucket='my-terraform-states',
        key_prefix='infrastructure'
    )

    # 初始化Terraform管理器
    tf_manager = TerraformManager(
        working_dir='./terraform',
        variables=variables
    )

    try:
        # 备份现有状态
        if args.action in ['apply', 'destroy']:
            state_manager.backup_state(args.env)

        # 执行操作
        if args.action == 'plan':
            output = tf_manager.plan()
        elif args.action == 'apply':
            output = tf_manager.apply()
        elif args.action == 'destroy':
            output = tf_manager.destroy()

        print(output)

    except Exception as e:
        print(f"Error: {str(e)}")
        raise

if __name__ == "__main__":
    main()

四、最佳实践建议

1. 错误处理和日志

import logging

def setup_logger():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    return logging.getLogger(__name__)

logger = setup_logger()

2. 环境变量管理

import os
from dotenv import load_dotenv

def setup_environment(env: str):
    """设置环境变量"""
    load_dotenv(f'.env.{env}')
    
    # 设置AWS凭证
    os.environ['AWS_PROFILE'] = os.getenv('AWS_PROFILE', 'default')
    os.environ['AWS_REGION'] = os.getenv('AWS_REGION', 'us-west-2')

3. 执行锁定机制

import fcntl
import contextlib

@contextlib.contextmanager
def deployment_lock(lock_file: str):
    with open(lock_file, 'w') as f:
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
            yield
        finally:
            fcntl.flock(f, fcntl.LOCK_UN)

五、使用示例

# 执行计划
python deploy.py --env dev --action plan

# 应用更改
python deploy.py --env dev --action apply

# 销毁资源
python deploy.py --env dev --action destroy

这种集成方式可以让你:

1. 通过Python控制Terraform的执行流程
2. 实现更复杂的配置管理
3. 添加额外的安全检查和验证
4. 集成到更大的自动化系统中
5. 实现更好的错误处理和日志记录
6. 提供更灵活的变量管理

记住要根据具体需求调整和扩展这些代码，确保适合你的使用场景。