Infrastructure as Code with Python

1import subprocess
2import json
3import os
4from pathlib import Path
5from typing import Dict, List, Optional, Any
6import logging
7from datetime import datetime
8
9logging.basicConfig(level=logging.INFO)
10logger = logging.getLogger(__name__)
11
12
13class TerraformRunner:
14    """
15    Production-grade Terraform orchestration with Python
16    """
17
18    def __init__(
19        self,
20        working_dir: Path,
21        terraform_bin: str = 'terraform',
22        var_file: Optional[Path] = None
23    ):
24        """
25        Initialize Terraform runner
26
27        Args:
28            working_dir: Directory containing Terraform configuration
29            terraform_bin: Path to terraform binary
30            var_file: Optional variables file path
31        """
32        self.working_dir = Path(working_dir)
33        self.terraform_bin = terraform_bin
34        self.var_file = var_file
35
36        # Verify directory exists
37        if not self.working_dir.exists():
38            raise ValueError(f"Working directory does not exist: {self.working_dir}")
39
40        logger.info(f"Terraform runner initialized: {self.working_dir}")
41
42    def _run_command(
43        self,
44        args: List[str],
45        capture_output: bool = True,
46        check: bool = True,
47        env: Optional[Dict] = None
48    ) -> subprocess.CompletedProcess:
49        """
50        Execute Terraform command
51
52        Args:
53            args: Command arguments
54            capture_output: Whether to capture stdout/stderr
55            check: Whether to raise exception on non-zero exit
56            env: Environment variables
57
58        Returns:
59            CompletedProcess instance
60        """
61        # Build full command
62        cmd = [self.terraform_bin] + args
63
64        # Set environment
65        command_env = os.environ.copy()
66        if env:
67            command_env.update(env)
68
69        logger.info(f"Executing: {' '.join(cmd)}")
70
71        try:
72            result = subprocess.run(
73                cmd,
74                cwd=self.working_dir,
75                capture_output=capture_output,
76                text=True,
77                check=check,
78                env=command_env
79            )
80
81            if result.stdout:
82                logger.debug(f"STDOUT: {result.stdout[:200]}...")
83
84            return result
85
86        except subprocess.CalledProcessError as e:
87            logger.error(f"Command failed: {' '.join(cmd)}")
88            logger.error(f"Exit code: {e.returncode}")
89            logger.error(f"STDERR: {e.stderr}")
90            raise
91
92    def init(
93        self,
94        backend_config: Optional[Dict[str, str]] = None,
95        upgrade: bool = False,
96        reconfigure: bool = False
97    ) -> Dict:
98        """
99        Initialize Terraform working directory
100
101        Args:
102            backend_config: Backend configuration key-value pairs
103            upgrade: Upgrade modules and plugins
104            reconfigure: Reconfigure backend
105
106        Returns:
107            Initialization result
108        """
109        args = ['init']
110
111        if upgrade:
112            args.append('-upgrade')
113
114        if reconfigure:
115            args.append('-reconfigure')
116
117        if backend_config:
118            for key, value in backend_config.items():
119                args.extend(['-backend-config', f'{key}={value}'])
120
121        start_time = datetime.now()
122        result = self._run_command(args)
123        duration = (datetime.now() - start_time).total_seconds()
124
125        logger.info(f"✓ Terraform init completed in {duration:.2f}s")
126
127        return {
128            'success': result.returncode == 0,
129            'duration': duration,
130            'output': result.stdout
131        }
132
133    def validate(self) -> Dict:
134        """
135        Validate Terraform configuration
136
137        Returns:
138            Validation result
139        """
140        args = ['validate', '-json']
141
142        result = self._run_command(args)
143
144        try:
145            validation_result = json.loads(result.stdout)
146        except json.JSONDecodeError:
147            validation_result = {'valid': False, 'error': 'Failed to parse JSON'}
148
149        if validation_result.get('valid'):
150            logger.info("✓ Configuration is valid")
151        else:
152            logger.error(f"✗ Configuration invalid: {validation_result.get('error_message')}")
153
154        return validation_result
155
156    def plan(
157        self,
158        out_file: Optional[Path] = None,
159        var_file: Optional[Path] = None,
160        variables: Optional[Dict[str, Any]] = None,
161        destroy: bool = False
162    ) -> Dict:
163        """
164        Generate and show execution plan
165
166        Args:
167            out_file: Save plan to file
168            var_file: Variable file path
169            variables: Variable key-value pairs
170            destroy: Generate destroy plan
171
172        Returns:
173            Plan result
174        """
175        args = ['plan', '-json']
176
177        if destroy:
178            args.append('-destroy')
179
180        if out_file:
181            args.extend(['-out', str(out_file)])
182
183        # Add variable file
184        var_file = var_file or self.var_file
185        if var_file:
186            args.extend(['-var-file', str(var_file)])
187
188        # Add inline variables
189        if variables:
190            for key, value in variables.items():
191                args.extend(['-var', f'{key}={value}'])
192
193        start_time = datetime.now()
194        result = self._run_command(args)
195        duration = (datetime.now() - start_time).total_seconds()
196
197        # Parse plan output
198        plan_summary = self._parse_plan_output(result.stdout)
199
200        logger.info(f"✓ Plan completed in {duration:.2f}s")
201        logger.info(f"  Add: {plan_summary['to_add']}, "
202                   f"Change: {plan_summary['to_change']}, "
203                   f"Destroy: {plan_summary['to_destroy']}")
204
205        return {
206            'success': True,
207            'duration': duration,
208            'summary': plan_summary,
209            'output': result.stdout
210        }
211
212    def _parse_plan_output(self, json_output: str) -> Dict:
213        """
214        Parse Terraform plan JSON output
215
216        Args:
217            json_output: JSON output from plan command
218
219        Returns:
220            Plan summary
221        """
222        summary = {
223            'to_add': 0,
224            'to_change': 0,
225            'to_destroy': 0,
226            'resources': []
227        }
228
229        for line in json_output.split('\n'):
230            if not line.strip():
231                continue
232
233            try:
234                data = json.loads(line)
235
236                if data.get('type') == 'planned_change':
237                    resource = data.get('change', {}).get('resource', {})
238                    action = data.get('change', {}).get('action')
239
240                    summary['resources'].append({
241                        'address': resource.get('addr'),
242                        'action': action
243                    })
244
245                    if action == 'create':
246                        summary['to_add'] += 1
247                    elif action == 'update':
248                        summary['to_change'] += 1
249                    elif action == 'delete':
250                        summary['to_destroy'] += 1
251
252            except json.JSONDecodeError:
253                continue
254
255        return summary
256
257    def apply(
258        self,
259        plan_file: Optional[Path] = None,
260        auto_approve: bool = False,
261        variables: Optional[Dict[str, Any]] = None
262    ) -> Dict:
263        """
264        Apply Terraform changes
265
266        Args:
267            plan_file: Path to saved plan file
268            auto_approve: Skip interactive approval
269            variables: Variable key-value pairs
270
271        Returns:
272            Apply result
273        """
274        args = ['apply']
275
276        if auto_approve:
277            args.append('-auto-approve')
278
279        if plan_file:
280            args.append(str(plan_file))
281        else:
282            # Add variables if not using plan file
283            if variables:
284                for key, value in variables.items():
285                    args.extend(['-var', f'{key}={value}'])
286
287        start_time = datetime.now()
288        result = self._run_command(args)
289        duration = (datetime.now() - start_time).total_seconds()
290
291        logger.info(f"✓ Apply completed in {duration:.2f}s")
292
293        return {
294            'success': result.returncode == 0,
295            'duration': duration,
296            'output': result.stdout
297        }
298
299    def destroy(
300        self,
301        auto_approve: bool = False,
302        target: Optional[List[str]] = None
303    ) -> Dict:
304        """
305        Destroy Terraform-managed infrastructure
306
307        Args:
308            auto_approve: Skip interactive approval
309            target: Specific resources to destroy
310
311        Returns:
312            Destroy result
313        """
314        args = ['destroy']
315
316        if auto_approve:
317            args.append('-auto-approve')
318
319        if target:
320            for t in target:
321                args.extend(['-target', t])
322
323        start_time = datetime.now()
324        result = self._run_command(args)
325        duration = (datetime.now() - start_time).total_seconds()
326
327        logger.info(f"✓ Destroy completed in {duration:.2f}s")
328
329        return {
330            'success': result.returncode == 0,
331            'duration': duration,
332            'output': result.stdout
333        }
334
335    def output(
336        self,
337        name: Optional[str] = None,
338        json_format: bool = True
339    ) -> Dict:
340        """
341        Read Terraform outputs
342
343        Args:
344            name: Specific output to retrieve
345            json_format: Return as JSON
346
347        Returns:
348            Output values
349        """
350        args = ['output']
351
352        if json_format:
353            args.append('-json')
354
355        if name:
356            args.append(name)
357
358        result = self._run_command(args)
359
360        if json_format:
361            try:
362                outputs = json.loads(result.stdout)
363                return outputs
364            except json.JSONDecodeError:
365                return {}
366        else:
367            return {'raw': result.stdout}
368
369    def state_list(self) -> List[str]:
370        """
371        List resources in Terraform state
372
373        Returns:
374            List of resource addresses
375        """
376        args = ['state', 'list']
377
378        result = self._run_command(args)
379        resources = [line.strip() for line in result.stdout.split('\n') if line.strip()]
380
381        logger.info(f"Found {len(resources)} resources in state")
382
383        return resources
384
385    def state_show(self, address: str) -> Dict:
386        """
387        Show details of a specific resource
388
389        Args:
390            address: Resource address
391
392        Returns:
393            Resource details
394        """
395        args = ['state', 'show', '-json', address]
396
397        result = self._run_command(args)
398
399        try:
400            return json.loads(result.stdout)
401        except json.JSONDecodeError:
402            return {}
403
404    def import_resource(
405        self,
406        address: str,
407        id: str
408    ) -> Dict:
409        """
410        Import existing infrastructure into Terraform state
411
412        Args:
413            address: Resource address in Terraform
414            id: Resource ID in provider
415
416        Returns:
417            Import result
418        """
419        args = ['import', address, id]
420
421        start_time = datetime.now()
422        result = self._run_command(args)
423        duration = (datetime.now() - start_time).total_seconds()
424
425        logger.info(f"✓ Imported {address} in {duration:.2f}s")
426
427        return {
428            'success': result.returncode == 0,
429            'duration': duration,
430            'output': result.stdout
431        }
432
433
434# Usage example
435terraform = TerraformRunner(
436    working_dir=Path('./terraform'),
437    var_file=Path('./terraform.tfvars')
438)
439
440# Initialize
441init_result = terraform.init(
442    backend_config={
443        'bucket': 'my-terraform-state',
444        'key': 'production/terraform.tfstate',
445        'region': 'us-east-1'
446    },
447    upgrade=True
448)
449
450# Validate configuration
451validation = terraform.validate()
452
453if validation.get('valid'):
454    # Generate plan
455    plan_result = terraform.plan(
456        out_file=Path('./tfplan'),
457        variables={
458            'environment': 'production',
459            'instance_count': '3'
460        }
461    )
462
463    print(f"Plan Summary:")
464    print(f"  Resources to add: {plan_result['summary']['to_add']}")
465    print(f"  Resources to change: {plan_result['summary']['to_change']}")
466    print(f"  Resources to destroy: {plan_result['summary']['to_destroy']}")
467
468    # Apply changes
469    apply_result = terraform.apply(
470        plan_file=Path('./tfplan'),
471        auto_approve=True
472    )
473
474    # Get outputs
475    outputs = terraform.output()
476    print(f"\nOutputs: {json.dumps(outputs, indent=2)}")
477
478    # List state resources
479    resources = terraform.state_list()
480    print(f"\nManaged Resources ({len(resources)}):")
481    for resource in resources[:5]:
482        print(f"  - {resource}")

1import hashlib
2from typing import Set
3import difflib
4
5
6class TerraformOrchestrator:
7    """
8    Advanced Terraform orchestration with drift detection
9    """
10
11    def __init__(self, working_dir: Path):
12        """
13        Initialize orchestrator
14
15        Args:
16            working_dir: Terraform working directory
17        """
18        self.terraform = TerraformRunner(working_dir)
19        self.state_cache = {}
20
21    def detect_drift(self) -> Dict:
22        """
23        Detect configuration drift between code and deployed infrastructure
24
25        Returns:
26            Drift detection results
27        """
28        logger.info("Detecting configuration drift...")
29
30        # Generate refresh-only plan
31        plan_result = self.terraform.plan()
32
33        drift_detected = (
34            plan_result['summary']['to_change'] > 0 or
35            plan_result['summary']['to_destroy'] > 0
36        )
37
38        return {
39            'drift_detected': drift_detected,
40            'changes_required': plan_result['summary']['to_change'],
41            'resources_out_of_sync': plan_result['summary']['to_destroy'],
42            'details': plan_result['summary']['resources']
43        }
44
45    def validate_before_apply(self) -> bool:
46        """
47        Comprehensive validation before applying changes
48
49        Returns:
50            Whether validation passed
51        """
52        logger.info("Running pre-apply validation...")
53
54        # 1. Validate syntax
55        validation = self.terraform.validate()
56        if not validation.get('valid'):
57            logger.error("Syntax validation failed")
58            return False
59
60        # 2. Generate plan
61        plan_result = self.terraform.plan()
62
63        # 3. Check for risky operations
64        for resource in plan_result['summary']['resources']:
65            if resource['action'] == 'delete':
66                logger.warning(f"DESTRUCTIVE: {resource['address']} will be deleted")
67
68        # 4. Verify no unexpected deletions
69        destroy_count = plan_result['summary']['to_destroy']
70        if destroy_count > 0:
71            logger.warning(f"{destroy_count} resources will be destroyed")
72            # In production, might require additional approval
73            return False
74
75        logger.info("✓ Pre-apply validation passed")
76        return True
77
78    def safe_apply(
79        self,
80        dry_run: bool = False,
81        require_approval: bool = True
82    ) -> Dict:
83        """
84        Safely apply Terraform changes with validation
85
86        Args:
87            dry_run: Only generate plan, don't apply
88            require_approval: Require manual approval
89
90        Returns:
91            Apply result
92        """
93        # Validate first
94        if not self.validate_before_apply():
95            return {
96                'success': False,
97                'error': 'Pre-apply validation failed'
98            }
99
100        # Generate plan
101        plan_result = self.terraform.plan(out_file=Path('./tfplan'))
102
103        if dry_run:
104            return {
105                'success': True,
106                'dry_run': True,
107                'plan': plan_result
108            }
109
110        # Require approval for changes
111        if require_approval and plan_result['summary']['to_change'] > 0:
112            logger.info("Changes require approval (use require_approval=False to bypass)")
113            return {
114                'success': False,
115                'approval_required': True,
116                'plan': plan_result
117            }
118
119        # Apply changes
120        apply_result = self.terraform.apply(
121            plan_file=Path('./tfplan'),
122            auto_approve=True
123        )
124
125        return {
126            'success': apply_result['success'],
127            'plan': plan_result,
128            'apply': apply_result
129        }
130
131    def backup_state(self, backup_path: Path):
132        """
133        Backup Terraform state file
134
135        Args:
136            backup_path: Path to save backup
137        """
138        state_file = self.terraform.working_dir / 'terraform.tfstate'
139
140        if state_file.exists():
141            import shutil
142            shutil.copy2(state_file, backup_path)
143            logger.info(f"State backed up to: {backup_path}")
144
145
146# Usage
147orchestrator = TerraformOrchestrator(working_dir=Path('./terraform'))
148
149# Detect drift
150drift_result = orchestrator.detect_drift()
151if drift_result['drift_detected']:
152    print(f"⚠️  Configuration drift detected!")
153    print(f"  Changes required: {drift_result['changes_required']}")
154
155# Safe apply with validation
156result = orchestrator.safe_apply(
157    dry_run=False,
158    require_approval=False
159)
160
161if result['success']:
162    print("✓ Infrastructure updated successfully")
163else:
164    print(f"✗ Apply failed: {result.get('error')}")

1import boto3
2from botocore.exceptions import ClientError
3from typing import Dict, List, Optional
4import time
5import logging
6
7logger = logging.getLogger(__name__)
8
9
10class AWSResourceProvisioner:
11    """
12    Provision and manage AWS resources using Boto3
13    """
14
15    def __init__(
16        self,
17        region: str = 'us-east-1',
18        profile: Optional[str] = None
19    ):
20        """
21        Initialize AWS provisioner
22
23        Args:
24            region: AWS region
25            profile: AWS credentials profile name
26        """
27        self.region = region
28
29        # Create session
30        if profile:
31            self.session = boto3.Session(
32                profile_name=profile,
33                region_name=region
34            )
35        else:
36            self.session = boto3.Session(region_name=region)
37
38        # Initialize clients
39        self.ec2_client = self.session.client('ec2')
40        self.ec2_resource = self.session.resource('ec2')
41        self.s3_client = self.session.client('s3')
42        self.rds_client = self.session.client('rds')
43        self.lambda_client = self.session.client('lambda')
44
45        logger.info(f"AWS provisioner initialized: {region}")
46
47    def create_vpc(
48        self,
49        cidr_block: str = '10.0.0.0/16',
50        name: Optional[str] = None
51    ) -> Dict:
52        """
53        Create VPC with CIDR block
54
55        Args:
56            cidr_block: VPC CIDR block
57            name: VPC name tag
58
59        Returns:
60            VPC details
61        """
62        try:
63            # Create VPC
64            vpc = self.ec2_resource.create_vpc(CidrBlock=cidr_block)
65            vpc.wait_until_available()
66
67            # Add name tag
68            if name:
69                vpc.create_tags(Tags=[{'Key': 'Name', 'Value': name}])
70
71            # Enable DNS
72            vpc.modify_attribute(EnableDnsSupport={'Value': True})
73            vpc.modify_attribute(EnableDnsHostnames={'Value': True})
74
75            logger.info(f"✓ VPC created: {vpc.id}")
76
77            return {
78                'vpc_id': vpc.id,
79                'cidr_block': cidr_block,
80                'state': vpc.state
81            }
82
83        except ClientError as e:
84            logger.error(f"Failed to create VPC: {e}")
85            raise
86
87    def create_subnet(
88        self,
89        vpc_id: str,
90        cidr_block: str,
91        availability_zone: str,
92        name: Optional[str] = None,
93        public: bool = False
94    ) -> Dict:
95        """
96        Create subnet in VPC
97
98        Args:
99            vpc_id: VPC ID
100            cidr_block: Subnet CIDR block
101            availability_zone: AZ for subnet
102            name: Subnet name tag
103            public: Whether subnet should auto-assign public IPs
104
105        Returns:
106            Subnet details
107        """
108        try:
109            # Create subnet
110            subnet = self.ec2_resource.create_subnet(
111                VpcId=vpc_id,
112                CidrBlock=cidr_block,
113                AvailabilityZone=availability_zone
114            )
115
116            # Configure public IP auto-assignment
117            if public:
118                self.ec2_client.modify_subnet_attribute(
119                    SubnetId=subnet.id,
120                    MapPublicIpOnLaunch={'Value': True}
121                )
122
123            # Add name tag
124            if name:
125                subnet.create_tags(Tags=[{'Key': 'Name', 'Value': name}])
126
127            logger.info(f"✓ Subnet created: {subnet.id}")
128
129            return {
130                'subnet_id': subnet.id,
131                'vpc_id': vpc_id,
132                'cidr_block': cidr_block,
133                'availability_zone': availability_zone,
134                'public': public
135            }
136
137        except ClientError as e:
138            logger.error(f"Failed to create subnet: {e}")
139            raise
140
141    def create_security_group(
142        self,
143        vpc_id: str,
144        name: str,
145        description: str,
146        ingress_rules: List[Dict]
147    ) -> Dict:
148        """
149        Create security group with ingress rules
150
151        Args:
152            vpc_id: VPC ID
153            name: Security group name
154            description: Security group description
155            ingress_rules: List of ingress rule dictionaries
156
157        Returns:
158            Security group details
159        """
160        try:
161            # Create security group
162            response = self.ec2_client.create_security_group(
163                GroupName=name,
164                Description=description,
165                VpcId=vpc_id
166            )
167
168            sg_id = response['GroupId']
169
170            # Add ingress rules
171            if ingress_rules:
172                self.ec2_client.authorize_security_group_ingress(
173                    GroupId=sg_id,
174                    IpPermissions=ingress_rules
175                )
176
177            logger.info(f"✓ Security group created: {sg_id}")
178
179            return {
180                'security_group_id': sg_id,
181                'name': name,
182                'vpc_id': vpc_id,
183                'ingress_rules': len(ingress_rules)
184            }
185
186        except ClientError as e:
187            logger.error(f"Failed to create security group: {e}")
188            raise
189
190    def launch_ec2_instance(
191        self,
192        ami_id: str,
193        instance_type: str = 't2.micro',
194        subnet_id: Optional[str] = None,
195        security_group_ids: Optional[List[str]] = None,
196        key_name: Optional[str] = None,
197        user_data: Optional[str] = None,
198        tags: Optional[Dict[str, str]] = None
199    ) -> Dict:
200        """
201        Launch EC2 instance
202
203        Args:
204            ami_id: AMI ID to launch
205            instance_type: EC2 instance type
206            subnet_id: Subnet ID
207            security_group_ids: Security group IDs
208            key_name: SSH key pair name
209            user_data: User data script
210            tags: Instance tags
211
212        Returns:
213            Instance details
214        """
215        try:
216            # Build launch parameters
217            launch_params = {
218                'ImageId': ami_id,
219                'InstanceType': instance_type,
220                'MinCount': 1,
221                'MaxCount': 1
222            }
223
224            if subnet_id:
225                launch_params['SubnetId'] = subnet_id
226
227            if security_group_ids:
228                launch_params['SecurityGroupIds'] = security_group_ids
229
230            if key_name:
231                launch_params['KeyName'] = key_name
232
233            if user_data:
234                launch_params['UserData'] = user_data
235
236            # Launch instance
237            instances = self.ec2_resource.create_instances(**launch_params)
238            instance = instances[0]
239
240            # Wait for instance to be running
241            instance.wait_until_running()
242            instance.reload()
243
244            # Add tags
245            if tags:
246                tag_list = [{'Key': k, 'Value': v} for k, v in tags.items()]
247                instance.create_tags(Tags=tag_list)
248
249            logger.info(f"✓ EC2 instance launched: {instance.id}")
250
251            return {
252                'instance_id': instance.id,
253                'instance_type': instance_type,
254                'state': instance.state['Name'],
255                'public_ip': instance.public_ip_address,
256                'private_ip': instance.private_ip_address
257            }
258
259        except ClientError as e:
260            logger.error(f"Failed to launch EC2 instance: {e}")
261            raise
262
263    def create_s3_bucket(
264        self,
265        bucket_name: str,
266        versioning: bool = False,
267        encryption: bool = True
268    ) -> Dict:
269        """
270        Create S3 bucket with configuration
271
272        Args:
273            bucket_name: Bucket name (must be globally unique)
274            versioning: Enable versioning
275            encryption: Enable server-side encryption
276
277        Returns:
278            Bucket details
279        """
280        try:
281            # Create bucket
282            if self.region == 'us-east-1':
283                self.s3_client.create_bucket(Bucket=bucket_name)
284            else:
285                self.s3_client.create_bucket(
286                    Bucket=bucket_name,
287                    CreateBucketConfiguration={'LocationConstraint': self.region}
288                )
289
290            # Enable versioning
291            if versioning:
292                self.s3_client.put_bucket_versioning(
293                    Bucket=bucket_name,
294                    VersioningConfiguration={'Status': 'Enabled'}
295                )
296
297            # Enable encryption
298            if encryption:
299                self.s3_client.put_bucket_encryption(
300                    Bucket=bucket_name,
301                    ServerSideEncryptionConfiguration={
302                        'Rules': [{
303                            'ApplyServerSideEncryptionByDefault': {
304                                'SSEAlgorithm': 'AES256'
305                            }
306                        }]
307                    }
308                )
309
310            logger.info(f"✓ S3 bucket created: {bucket_name}")
311
312            return {
313                'bucket_name': bucket_name,
314                'region': self.region,
315                'versioning': versioning,
316                'encryption': encryption
317            }
318
319        except ClientError as e:
320            logger.error(f"Failed to create S3 bucket: {e}")
321            raise
322
323    def create_rds_instance(
324        self,
325        db_instance_id: str,
326        db_engine: str = 'postgres',
327        db_instance_class: str = 'db.t3.micro',
328        allocated_storage: int = 20,
329        master_username: str = 'admin',
330        master_password: str = None,
331        db_name: Optional[str] = None
332    ) -> Dict:
333        """
334        Create RDS database instance
335
336        Args:
337            db_instance_id: Database instance identifier
338            db_engine: Database engine (postgres, mysql, etc.)
339            db_instance_class: Instance class
340            allocated_storage: Storage in GB
341            master_username: Master username
342            master_password: Master password
343            db_name: Initial database name
344
345        Returns:
346            RDS instance details
347        """
348        try:
349            # Create RDS instance
350            response = self.rds_client.create_db_instance(
351                DBInstanceIdentifier=db_instance_id,
352                Engine=db_engine,
353                DBInstanceClass=db_instance_class,
354                AllocatedStorage=allocated_storage,
355                MasterUsername=master_username,
356                MasterUserPassword=master_password,
357                DBName=db_name
358            )
359
360            logger.info(f"✓ RDS instance creation initiated: {db_instance_id}")
361
362            db_instance = response['DBInstance']
363
364            return {
365                'db_instance_id': db_instance_id,
366                'engine': db_engine,
367                'instance_class': db_instance_class,
368                'status': db_instance['DBInstanceStatus'],
369                'endpoint': db_instance.get('Endpoint', {}).get('Address')
370            }
371
372        except ClientError as e:
373            logger.error(f"Failed to create RDS instance: {e}")
374            raise
375
376    def list_resources(self) -> Dict:
377        """
378        List all provisioned resources
379
380        Returns:
381            Resource inventory
382        """
383        resources = {
384            'vpcs': [],
385            'ec2_instances': [],
386            's3_buckets': [],
387            'rds_instances': []
388        }
389
390        try:
391            # List VPCs
392            vpcs = self.ec2_client.describe_vpcs()
393            resources['vpcs'] = [
394                {'vpc_id': vpc['VpcId'], 'cidr': vpc['CidrBlock']}
395                for vpc in vpcs['Vpcs']
396            ]
397
398            # List EC2 instances
399            instances = self.ec2_client.describe_instances()
400            for reservation in instances['Reservations']:
401                for instance in reservation['Instances']:
402                    resources['ec2_instances'].append({
403                        'instance_id': instance['InstanceId'],
404                        'type': instance['InstanceType'],
405                        'state': instance['State']['Name']
406                    })
407
408            # List S3 buckets
409            buckets = self.s3_client.list_buckets()
410            resources['s3_buckets'] = [
411                {'name': bucket['Name']}
412                for bucket in buckets['Buckets']
413            ]
414
415            # List RDS instances
416            db_instances = self.rds_client.describe_db_instances()
417            resources['rds_instances'] = [
418                {
419                    'db_instance_id': db['DBInstanceIdentifier'],
420                    'engine': db['Engine'],
421                    'status': db['DBInstanceStatus']
422                }
423                for db in db_instances['DBInstances']
424            ]
425
426            logger.info("✓ Resource inventory completed")
427
428        except ClientError as e:
429            logger.error(f"Failed to list resources: {e}")
430
431        return resources
432
433
434# Usage example
435provisioner = AWSResourceProvisioner(region='us-east-1')
436
437# Create VPC
438vpc = provisioner.create_vpc(
439    cidr_block='10.0.0.0/16',
440    name='production-vpc'
441)
442
443# Create subnet
444subnet = provisioner.create_subnet(
445    vpc_id=vpc['vpc_id'],
446    cidr_block='10.0.1.0/24',
447    availability_zone='us-east-1a',
448    name='public-subnet-1a',
449    public=True
450)
451
452# Create security group
453sg = provisioner.create_security_group(
454    vpc_id=vpc['vpc_id'],
455    name='web-sg',
456    description='Security group for web servers',
457    ingress_rules=[
458        {
459            'IpProtocol': 'tcp',
460            'FromPort': 80,
461            'ToPort': 80,
462            'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
463        },
464        {
465            'IpProtocol': 'tcp',
466            'FromPort': 443,
467            'ToPort': 443,
468            'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
469        }
470    ]
471)
472
473# Launch EC2 instance
474instance = provisioner.launch_ec2_instance(
475    ami_id='ami-0c55b159cbfafe1f0',
476    instance_type='t2.micro',
477    subnet_id=subnet['subnet_id'],
478    security_group_ids=[sg['security_group_id']],
479    tags={'Name': 'web-server-1', 'Environment': 'production'}
480)
481
482print(f"Instance launched: {instance['instance_id']}")
483print(f"Public IP: {instance['public_ip']}")
484
485# Create S3 bucket
486bucket = provisioner.create_s3_bucket(
487    bucket_name='my-app-data-bucket-12345',
488    versioning=True,
489    encryption=True
490)
491
492# List all resources
493resources = provisioner.list_resources()
494print(f"\nResource Inventory:")
495print(f"  VPCs: {len(resources['vpcs'])}")
496print(f"  EC2 Instances: {len(resources['ec2_instances'])}")
497print(f"  S3 Buckets: {len(resources['s3_buckets'])}")
498print(f"  RDS Instances: {len(resources['rds_instances'])}")

1import json
2from pathlib import Path
3from typing import Dict, List, Optional, Any
4from datetime import datetime
5import hashlib
6import copy
7
8
9class InfrastructureStateManager:
10    """
11    Manage infrastructure state with version control and drift detection
12    """
13
14    def __init__(self, state_file: Path = Path('./infrastructure.state')):
15        """
16        Initialize state manager
17
18        Args:
19            state_file: Path to state file
20        """
21        self.state_file = state_file
22        self.state = self._load_state()
23        self.history = []
24
25    def _load_state(self) -> Dict:
26        """Load state from file"""
27        if self.state_file.exists():
28            try:
29                with open(self.state_file, 'r') as f:
30                    return json.load(f)
31            except:
32                pass
33
34        return {
35            'version': 1,
36            'created_at': datetime.now().isoformat(),
37            'updated_at': datetime.now().isoformat(),
38            'resources': {},
39            'outputs': {}
40        }
41
42    def _save_state(self):
43        """Save state to file"""
44        self.state['updated_at'] = datetime.now().isoformat()
45
46        # Create backup
47        if self.state_file.exists():
48            backup_file = Path(str(self.state_file) + '.backup')
49            import shutil
50            shutil.copy2(self.state_file, backup_file)
51
52        # Save current state
53        with open(self.state_file, 'w') as f:
54            json.dump(self.state, f, indent=2)
55
56    def add_resource(
57        self,
58        resource_type: str,
59        resource_id: str,
60        attributes: Dict[str, Any]
61    ):
62        """
63        Add resource to state
64
65        Args:
66            resource_type: Type of resource (vpc, ec2_instance, etc.)
67            resource_id: Unique resource identifier
68            attributes: Resource attributes
69        """
70        resource_key = f"{resource_type}.{resource_id}"
71
72        self.state['resources'][resource_key] = {
73            'type': resource_type,
74            'id': resource_id,
75            'attributes': attributes,
76            'created_at': datetime.now().isoformat(),
77            'updated_at': datetime.now().isoformat(),
78            'checksum': self._calculate_checksum(attributes)
79        }
80
81        self._save_state()
82        logger.info(f"Added resource to state: {resource_key}")
83
84    def update_resource(
85        self,
86        resource_type: str,
87        resource_id: str,
88        attributes: Dict[str, Any]
89    ):
90        """
91        Update resource in state
92
93        Args:
94            resource_type: Type of resource
95            resource_id: Resource identifier
96            attributes: Updated attributes
97        """
98        resource_key = f"{resource_type}.{resource_id}"
99
100        if resource_key in self.state['resources']:
101            old_checksum = self.state['resources'][resource_key]['checksum']
102            new_checksum = self._calculate_checksum(attributes)
103
104            if old_checksum != new_checksum:
105                self.state['resources'][resource_key]['attributes'] = attributes
106                self.state['resources'][resource_key]['updated_at'] = datetime.now().isoformat()
107                self.state['resources'][resource_key]['checksum'] = new_checksum
108
109                self._save_state()
110                logger.info(f"Updated resource: {resource_key}")
111            else:
112                logger.debug(f"No changes detected: {resource_key}")
113        else:
114            logger.warning(f"Resource not found: {resource_key}")
115
116    def remove_resource(
117        self,
118        resource_type: str,
119        resource_id: str
120    ):
121        """
122        Remove resource from state
123
124        Args:
125            resource_type: Type of resource
126            resource_id: Resource identifier
127        """
128        resource_key = f"{resource_type}.{resource_id}"
129
130        if resource_key in self.state['resources']:
131            del self.state['resources'][resource_key]
132            self._save_state()
133            logger.info(f"Removed resource: {resource_key}")
134
135    def get_resource(
136        self,
137        resource_type: str,
138        resource_id: str
139    ) -> Optional[Dict]:
140        """
141        Get resource from state
142
143        Args:
144            resource_type: Type of resource
145            resource_id: Resource identifier
146
147        Returns:
148            Resource data or None
149        """
150        resource_key = f"{resource_type}.{resource_id}"
151        return self.state['resources'].get(resource_key)
152
153    def list_resources(
154        self,
155        resource_type: Optional[str] = None
156    ) -> List[Dict]:
157        """
158        List resources in state
159
160        Args:
161            resource_type: Optional filter by type
162
163        Returns:
164            List of resources
165        """
166        resources = []
167
168        for key, resource in self.state['resources'].items():
169            if resource_type is None or resource['type'] == resource_type:
170                resources.append({
171                    'key': key,
172                    **resource
173                })
174
175        return resources
176
177    def _calculate_checksum(self, data: Any) -> str:
178        """
179        Calculate checksum of data
180
181        Args:
182            data: Data to checksum
183
184        Returns:
185            SHA256 checksum
186        """
187        json_str = json.dumps(data, sort_keys=True)
188        return hashlib.sha256(json_str.encode()).hexdigest()
189
190    def detect_drift(self, current_state: Dict) -> Dict:
191        """
192        Detect drift between stored and current state
193
194        Args:
195            current_state: Current infrastructure state
196
197        Returns:
198            Drift detection results
199        """
200        drift_results = {
201            'drift_detected': False,
202            'added': [],
203            'modified': [],
204            'deleted': []
205        }
206
207        # Check for modified and deleted resources
208        for key, resource in self.state['resources'].items():
209            if key not in current_state:
210                drift_results['deleted'].append(key)
211                drift_results['drift_detected'] = True
212            else:
213                current_checksum = self._calculate_checksum(
214                    current_state[key]['attributes']
215                )
216                stored_checksum = resource['checksum']
217
218                if current_checksum != stored_checksum:
219                    drift_results['modified'].append({
220                        'resource': key,
221                        'changes': self._diff_attributes(
222                            resource['attributes'],
223                            current_state[key]['attributes']
224                        )
225                    })
226                    drift_results['drift_detected'] = True
227
228        # Check for added resources
229        for key in current_state:
230            if key not in self.state['resources']:
231                drift_results['added'].append(key)
232                drift_results['drift_detected'] = True
233
234        return drift_results
235
236    def _diff_attributes(
237        self,
238        old_attrs: Dict,
239        new_attrs: Dict
240    ) -> Dict:
241        """
242        Calculate difference between attribute dictionaries
243
244        Args:
245            old_attrs: Old attributes
246            new_attrs: New attributes
247
248        Returns:
249            Difference dictionary
250        """
251        changes = {}
252
253        all_keys = set(old_attrs.keys()) | set(new_attrs.keys())
254
255        for key in all_keys:
256            old_value = old_attrs.get(key)
257            new_value = new_attrs.get(key)
258
259            if old_value != new_value:
260                changes[key] = {
261                    'old': old_value,
262                    'new': new_value
263                }
264
265        return changes
266
267    def export_state(self) -> Dict:
268        """
269        Export state for sharing or backup
270
271        Returns:
272            State dictionary
273        """
274        return copy.deepcopy(self.state)
275
276    def import_state(self, state_data: Dict):
277        """
278        Import state from external source
279
280        Args:
281            state_data: State dictionary to import
282        """
283        self.state = state_data
284        self._save_state()
285        logger.info("State imported successfully")
286
287
288# Usage example
289state_mgr = InfrastructureStateManager()
290
291# Add resources
292state_mgr.add_resource(
293    resource_type='vpc',
294    resource_id='vpc-12345',
295    attributes={
296        'cidr_block': '10.0.0.0/16',
297        'enable_dns': True,
298        'region': 'us-east-1'
299    }
300)
301
302state_mgr.add_resource(
303    resource_type='ec2_instance',
304    resource_id='i-abcdef',
305    attributes={
306        'instance_type': 't2.micro',
307        'ami': 'ami-12345',
308        'vpc_id': 'vpc-12345'
309    }
310)
311
312# List resources
313vpc_resources = state_mgr.list_resources(resource_type='vpc')
314print(f"VPCs in state: {len(vpc_resources)}")
315
316# Simulate current infrastructure state
317current_infrastructure = {
318    'vpc.vpc-12345': {
319        'type': 'vpc',
320        'id': 'vpc-12345',
321        'attributes': {
322            'cidr_block': '10.0.0.0/16',
323            'enable_dns': False,  # Changed!
324            'region': 'us-east-1'
325        }
326    },
327    'ec2_instance.i-abcdef': {
328        'type': 'ec2_instance',
329        'id': 'i-abcdef',
330        'attributes': {
331            'instance_type': 't2.micro',
332            'ami': 'ami-12345',
333            'vpc_id': 'vpc-12345'
334        }
335    }
336}
337
338# Detect drift
339drift = state_mgr.detect_drift(current_infrastructure)
340
341if drift['drift_detected']:
342    print("\n⚠️  Configuration drift detected!")
343    print(f"Modified resources: {len(drift['modified'])}")
344    for mod in drift['modified']:
345        print(f"  - {mod['resource']}")
346        for key, change in mod['changes'].items():
347            print(f"    {key}: {change['old']} -> {change['new']}")