Introduction to Python Automation

1#!/usr/bin/env python3
2"""
3🚀 Production-Ready DevOps Automation Script
4Demonstrates: Monitoring, Logging, Error Handling, System Checks
5Author: DevOps Automation Team
6"""
7
8import os
9import sys
10import logging
11import subprocess
12import platform
13from datetime import datetime
14from pathlib import Path
15from typing import Dict, List, Tuple
16
17# Configure professional logging
18logging.basicConfig(
19    level=logging.INFO,
20    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
21    handlers=[
22        logging.FileHandler('automation.log'),
23        logging.StreamHandler(sys.stdout)
24    ]
25)
26logger = logging.getLogger(__name__)
27
28class SystemAutomation:
29    """
30    Professional system automation class demonstrating DevOps patterns
31    """
32
33    def __init__(self, environment: str = "development"):
34        """
35        Initialize automation with environment-specific settings
36
37        Args:
38            environment: Target environment (development/staging/production)
39        """
40        self.environment = environment
41        self.platform = platform.system()
42        self.start_time = datetime.now()
43
44        logger.info(f"🚀 Initializing automation for {environment} environment")
45        logger.info(f"📊 Platform: {self.platform}")
46
47    def check_prerequisites(self) -> bool:
48        """
49        Verify all prerequisites before running automation
50
51        Returns:
52            bool: True if all checks pass
53        """
54        logger.info("🔍 Checking prerequisites...")
55
56        checks = {
57            "Python Version": self._check_python_version(),
58            "Required Directories": self._check_directories(),
59            "Network Connectivity": self._check_network(),
60            "Disk Space": self._check_disk_space()
61        }
62
63        for check_name, result in checks.items():
64            status = "✅ PASS" if result else "❌ FAIL"
65            logger.info(f"  {check_name}: {status}")
66
67        return all(checks.values())
68
69    def _check_python_version(self) -> bool:
70        """Verify Python version meets requirements"""
71        major, minor = sys.version_info[:2]
72        required = (3, 8)
73
74        if (major, minor) >= required:
75            logger.info(f"    Python {major}.{minor} meets requirement (>= 3.8)")
76            return True
77        else:
78            logger.error(f"    Python {major}.{minor} below requirement (>= 3.8)")
79            return False
80
81    def _check_directories(self) -> bool:
82        """Create required directories if they don't exist"""
83        required_dirs = ['logs', 'backups', 'temp', 'configs']
84
85        try:
86            for dir_name in required_dirs:
87                Path(dir_name).mkdir(exist_ok=True)
88            logger.info(f"    Created/verified {len(required_dirs)} directories")
89            return True
90        except Exception as e:
91            logger.error(f"    Directory creation failed: {e}")
92            return False
93
94    def _check_network(self) -> bool:
95        """Verify network connectivity"""
96        try:
97            # Simple connectivity check
98            subprocess.run(
99                ['ping', '-c', '1', '8.8.8.8'] if self.platform != 'Windows' else ['ping', '-n', '1', '8.8.8.8'],
100                stdout=subprocess.DEVNULL,
101                stderr=subprocess.DEVNULL,
102                timeout=5
103            )
104            logger.info("    Network connectivity verified")
105            return True
106        except Exception:
107            logger.warning("    Network connectivity check failed (non-critical)")
108            return True  # Non-critical for this demo
109
110    def _check_disk_space(self) -> bool:
111        """Check available disk space"""
112        try:
113            import shutil
114            stat = shutil.disk_usage('.')
115            free_gb = stat.free / (1024**3)
116
117            if free_gb > 1:  # Require at least 1GB free
118                logger.info(f"    Disk space: {free_gb:.2f} GB available")
119                return True
120            else:
121                logger.error(f"    Insufficient disk space: {free_gb:.2f} GB")
122                return False
123        except Exception as e:
124            logger.error(f"    Disk space check failed: {e}")
125            return False
126
127    def execute_automation_tasks(self) -> Dict[str, bool]:
128        """
129        Execute main automation tasks with error handling
130
131        Returns:
132            Dict mapping task names to success status
133        """
134        logger.info("\n🔧 Executing automation tasks...")
135
136        tasks = {
137            "System Information": self._gather_system_info,
138            "Process Monitoring": self._monitor_processes,
139            "File Cleanup": self._cleanup_temp_files,
140            "Configuration Backup": self._backup_configurations,
141            "Health Check": self._perform_health_check
142        }
143
144        results = {}
145        for task_name, task_func in tasks.items():
146            try:
147                logger.info(f"\n▶️  Running: {task_name}")
148                success = task_func()
149                results[task_name] = success
150
151                if success:
152                    logger.info(f"  ✅ {task_name} completed successfully")
153                else:
154                    logger.warning(f"  ⚠️  {task_name} completed with warnings")
155
156            except Exception as e:
157                logger.error(f"  ❌ {task_name} failed: {e}")
158                results[task_name] = False
159
160        return results
161
162    def _gather_system_info(self) -> bool:
163        """Gather comprehensive system information"""
164        info = {
165            "Platform": platform.platform(),
166            "Architecture": platform.machine(),
167            "Python Version": platform.python_version(),
168            "Hostname": platform.node(),
169            "Processor": platform.processor()
170        }
171
172        logger.info("  📊 System Information:")
173        for key, value in info.items():
174            logger.info(f"    {key}: {value}")
175
176        return True
177
178    def _monitor_processes(self) -> bool:
179        """Monitor system processes (simplified)"""
180        try:
181            # Get process count
182            if self.platform == "Windows":
183                result = subprocess.run(['tasklist'], capture_output=True, text=True)
184                process_count = len(result.stdout.split('\n')) - 3
185            else:
186                result = subprocess.run(['ps', 'aux'], capture_output=True, text=True)
187                process_count = len(result.stdout.split('\n')) - 1
188
189            logger.info(f"  📈 Active processes: {process_count}")
190            return True
191        except Exception as e:
192            logger.error(f"  Process monitoring failed: {e}")
193            return False
194
195    def _cleanup_temp_files(self) -> bool:
196        """Clean up temporary files"""
197        temp_dir = Path('temp')
198        if temp_dir.exists():
199            file_count = len(list(temp_dir.glob('*')))
200            logger.info(f"  🧹 Found {file_count} temporary files")
201            # In production, would actually clean files here
202            return True
203        return True
204
205    def _backup_configurations(self) -> bool:
206        """Backup configuration files"""
207        backup_dir = Path('backups')
208        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
209        backup_path = backup_dir / f"config_backup_{timestamp}.txt"
210
211        try:
212            # Create a simple backup (in production, would backup actual configs)
213            with open(backup_path, 'w') as f:
214                f.write(f"Backup created at: {datetime.now()}\n")
215                f.write(f"Environment: {self.environment}\n")
216
217            logger.info(f"  💾 Backup created: {backup_path.name}")
218            return True
219        except Exception as e:
220            logger.error(f"  Backup failed: {e}")
221            return False
222
223    def _perform_health_check(self) -> bool:
224        """Perform system health check"""
225        health_metrics = {
226            "Automation Framework": "Operational",
227            "Error Handling": "Active",
228            "Logging System": "Functional",
229            "File System": "Accessible"
230        }
231
232        logger.info("  🏥 Health Check Results:")
233        for metric, status in health_metrics.items():
234            logger.info(f"    {metric}: {status}")
235
236        return True
237
238    def generate_report(self, results: Dict[str, bool]):
239        """
240        Generate automation execution report
241
242        Args:
243            results: Task execution results
244        """
245        logger.info("\n📋 AUTOMATION EXECUTION REPORT")
246        logger.info("=" * 60)
247
248        total_tasks = len(results)
249        successful_tasks = sum(results.values())
250        failed_tasks = total_tasks - successful_tasks
251        success_rate = (successful_tasks / total_tasks) * 100 if total_tasks > 0 else 0
252
253        logger.info(f"Environment: {self.environment}")
254        logger.info(f"Start Time: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}")
255        logger.info(f"End Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
256        logger.info(f"Duration: {(datetime.now() - self.start_time).total_seconds():.2f} seconds")
257        logger.info(f"\nTask Summary:")
258        logger.info(f"  Total Tasks: {total_tasks}")
259        logger.info(f"  Successful: {successful_tasks} ✅")
260        logger.info(f"  Failed: {failed_tasks} ❌")
261        logger.info(f"  Success Rate: {success_rate:.1f}%")
262
263        logger.info(f"\nDetailed Results:")
264        for task, success in results.items():
265            status = "✅ SUCCESS" if success else "❌ FAILED"
266            logger.info(f"  {task}: {status}")
267
268        logger.info("=" * 60)
269
270def main():
271    """
272    Main entry point for automation script
273    """
274    print("\n" + "="*60)
275    print("🚀 DevOps Automation Script - Production Ready")
276    print("="*60 + "\n")
277
278    try:
279        # Initialize automation
280        automation = SystemAutomation(environment="development")
281
282        # Check prerequisites
283        if not automation.check_prerequisites():
284            logger.error("❌ Prerequisites check failed. Aborting.")
285            sys.exit(1)
286
287        logger.info("✅ All prerequisites passed\n")
288
289        # Execute automation tasks
290        results = automation.execute_automation_tasks()
291
292        # Generate final report
293        automation.generate_report(results)
294
295        # Determine exit code based on results
296        if all(results.values()):
297            logger.info("\n🎉 Automation completed successfully!")
298            sys.exit(0)
299        else:
300            logger.warning("\n⚠️  Automation completed with some failures")
301            sys.exit(1)
302
303    except KeyboardInterrupt:
304        logger.info("\n⚠️  Automation interrupted by user")
305        sys.exit(130)
306    except Exception as e:
307        logger.error(f"\n❌ Fatal error: {e}")
308        sys.exit(1)
309
310if __name__ == "__main__":
311    main()