Python Fundamentals for DevOps: Your First Automation Scripts

# Check if Python is installed
python3 --version
# Should be 3.8 or higher

# Install Python on Ubuntu/Debian
sudo apt update
sudo apt install python3 python3-pip python3-venv

# Install Python on macOS
brew install python3

# Install Python on Windows (download from python.org)
# Or use WSL2 with Ubuntu

# Create a virtual environment
python3 -m venv myenv

# Activate it (Linux/macOS)
source myenv/bin/activate

# Activate it (Windows)
myenv\Scripts\activate

# You'll see (myenv) in your prompt
# Install packages safely
pip install requests boto3

# Exit virtual environment
deactivate

#!/usr/bin/env python3
# my_first_script.py - Your first Python automation script

# This is a comment - Python ignores this

# Variables (no $ sign like Bash!)
name = "DevOps Engineer"
version = 1.0

# Print to console (like echo in Bash)
print("Hello, World!")
print(f"Hello, {name}!")  # f-string = formatted string
print(f"Script version: {version}")

# Get user input
user_name = input("What is your name? ")
print(f"Nice to meet you, {user_name}!")

# Make script executable
chmod +x my_first_script.py

# Run it
./my_first_script.py
# OR
python3 my_first_script.py

#!/usr/bin/env python3
# data_types.py - Exploring Python data types

# String (text)
project_name = "Infrastructure Automation"
cloud_provider = 'AWS'  # Single quotes also work

# Integer (whole number)
cpu_cores = 8
memory_gb = 16

# Float (decimal number)
price_per_hour = 0.42
disk_usage_percent = 75.5

# Boolean (True/False)
is_production = True
has_backup = False

# List (ordered, changeable, duplicates allowed)
servers = ["web-01", "web-02", "db-01", "cache-01"]
ports = [80, 443, 22, 3306]

# Dictionary (key-value pairs - like a JSON object)
server_config = {
    "name": "web-01",
    "ip": "10.0.1.10",
    "region": "us-east-1",
    "instance_type": "t3.medium"
}

# Print with type checking
print(f"Project name is a {type(project_name)}")
print(f"CPU cores is a {type(cpu_cores)}")
print(f"Servers is a {type(servers)}")
print(f"Server config is a {type(server_config)}")

#!/usr/bin/env python3
# lists_demo.py - Managing collections of items

# Create a list
servers = ["web-01", "web-02", "web-03", "db-01"]

# Access items (0-indexed!)
first_server = servers[0]    # "web-01"
second_server = servers[1]   # "web-02"
last_server = servers[-1]    # "db-01" (negative counts from end)

# Add items
servers.append("cache-01")    # Add to end
servers.insert(0, "lb-01")    # Insert at beginning

# Remove items
servers.remove("web-03")       # Remove by value
removed = servers.pop()        # Remove and return last item

# Loop through list (MOST IMPORTANT PATTERN)
for server in servers:
    print(f"Checking {server}...")

# List comprehension (creating lists from existing data)
uptime_seconds = [100, 2500, 50000, 7200]
uptime_hours = [seconds // 3600 for seconds in uptime_seconds]
print(f"Uptime in hours: {uptime_hours}")

# Check if item exists
if "web-01" in servers:
    print("web-01 is in the list")

# Get list length
print(f"I have {len(servers)} servers")

#!/usr/bin/env python3
# dictionaries_demo.py - Configuration management

# Create a dictionary
server = {
    "name": "web-01",
    "ip": "10.0.1.10",
    "region": "us-east-1",
    "tags": ["production", "web", "frontend"]
}

# Access values
name = server["name"]           # "web-01"
ip = server.get("ip")            # "10.0.1.10" (safer - returns None if missing)
region = server.get("region", "unknown")  # Default value if missing

# Add or update values
server["instance_type"] = "t3.medium"
server["status"] = "running"

# Loop through dictionaries (VERY COMMON IN DEVOPS)
for key, value in server.items():
    print(f"{key}: {value}")

# Get all keys or values
all_keys = server.keys()
all_values = server.values()

# Nested dictionaries (realistic infrastructure)
infrastructure = {
    "web": {
        "count": 3,
        "instance_type": "t3.medium",
        "subnet": "public-subnet-01"
    },
    "db": {
        "count": 2,
        "instance_type": "r5.large",
        "subnet": "private-subnet-01"
    }
}

print(f"Web instance type: {infrastructure['web']['instance_type']}")

#!/usr/bin/env python3
# conditionals.py - Making decisions in code

# Basic if statement
disk_usage = 85

if disk_usage > 90:
    print("CRITICAL: Disk almost full!")
elif disk_usage > 80:
    print("WARNING: Disk usage is high")
elif disk_usage > 70:
    print("INFO: Disk usage is elevated")
else:
    print("OK: Disk usage is normal")

# Real DevOps example: Check if server needs attention
server_status = {
    "cpu": 95,
    "memory": 60,
    "disk": 45,
    "is_alive": True
}

if not server_status["is_alive"]:
    print("🚨 Server is DOWN!")
elif server_status["cpu"] > 90:
    print("⚠️ CPU critical - investigate immediately")
elif server_status["memory"] > 85:
    print("⚠️ Memory critical - consider scaling up")
else:
    print("✓ Server is healthy")

# Multiple conditions with 'and' and 'or'
if disk_usage > 80 and server_status["cpu"] > 80:
    print("Multiple metrics critical - system under heavy load")

# Check if value exists
if "web-01" in servers:
    print("Server found")

#!/usr/bin/env python3
# loops.py - Repeating actions

# For loop: iterate through a list
servers = ["web-01", "web-02", "db-01", "cache-01"]

for server in servers:
    print(f"Pinging {server}...")
    # In real life, you'd check SSH connection here

# For loop with range
for i in range(5):
    print(f"Attempt {i+1} of 5")

# While loop: repeat until condition changes
attempt = 1
max_attempts = 5
service_ready = False

while not service_ready and attempt <= max_attempts:
    print(f"Checking service (attempt {attempt})...")
    # In real life, you'd check HTTP endpoint
    if attempt == 3:
        service_ready = True
        print("✓ Service is ready!")
    attempt += 1

if not service_ready:
    print("✗ Service failed to start")

# Loop with break (exit early)
for server in servers:
    if server == "db-01":
        print(f"Found database server: {server}")
        break  # Stop searching

# Loop with continue (skip this iteration)
for server in servers:
    if server.startswith("web"):
        print(f"Web server: {server}")
        continue  # Skip the rest of the loop
    print(f"Non-web server: {server}")

#!/usr/bin/env python3
# functions.py - Writing reusable code

# Define a simple function
def greet(name):
    """Print a greeting"""
    print(f"Hello, {name}!")

# Call the function
greet("Alex")

# Function with return value
def calculate_disk_usage(used, total):
    """Calculate disk usage percentage"""
    percentage = (used / total) * 100
    return round(percentage, 2)

usage = calculate_disk_usage(45, 100)
print(f"Disk usage: {usage}%")

# Function with default parameters
def check_server(server_name, port=22, timeout=5):
    """Check if server is reachable"""
    print(f"Checking {server_name}:{port} (timeout: {timeout}s)")
    # In real life, you'd attempt connection
    return True

check_server("web-01")                    # Uses default port 22
check_server("db-01", port=3306)          # Custom port
check_server("cache-01", port=6379, timeout=10)

# Function with multiple returns
def get_server_stats(server_name):
    """Return multiple values as a tuple"""
    cpu = 45
    memory = 60
    disk = 35
    return cpu, memory, disk

cpu, memory, disk = get_server_stats("web-01")
print(f"CPU: {cpu}%, Memory: {memory}%, Disk: {disk}%")

# Real DevOps: Server health check function
def is_server_healthy(server):
    """
    Check if a server is healthy based on metrics
    Returns True if healthy, False otherwise
    """
    cpu = server.get("cpu", 0)
    memory = server.get("memory", 0)
    is_alive = server.get("is_alive", False)
    
    if not is_alive:
        return False
    if cpu > 90:
        return False
    if memory > 90:
        return False
    return True

# Test the function
server1 = {"name": "web-01", "cpu": 45, "memory": 60, "is_alive": True}
server2 = {"name": "web-02", "cpu": 95, "memory": 80, "is_alive": True}

print(f"Server1 healthy: {is_server_healthy(server1)}")
print(f"Server2 healthy: {is_server_healthy(server2)}")

#!/usr/bin/env python3
"""
disk_monitor.py - Monitor disk space on multiple servers
A practical DevOps automation script
"""

import subprocess
import json
from datetime import datetime

# Configuration
SERVERS = [
    {"name": "web-01", "ip": "10.0.1.10", "threshold": 80},
    {"name": "web-02", "ip": "10.0.1.11", "threshold": 80},
    {"name": "db-01", "ip": "10.0.2.10", "threshold": 75},
]

WARNING_THRESHOLD = 80
CRITICAL_THRESHOLD = 90

def get_disk_usage(server_ip):
    """
    Get disk usage percentage from a server via SSH
    Returns usage percentage as integer
    """
    try:
        # Command to get disk usage on / partition
        cmd = f"ssh -o ConnectTimeout=5 {server_ip} 'df -h / | tail -1 | awk '{{print $5}}' | sed 's/%//''"
        
        # Run command (simulated - in real life, you'd use paramiko)
        # For demo, we'll simulate response
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=10)
        
        if result.returncode == 0:
            usage = int(result.stdout.strip())
            return usage
        else:
            print(f"Error connecting to {server_ip}: {result.stderr}")
            return None
    except subprocess.TimeoutExpired:
        print(f"Timeout connecting to {server_ip}")
        return None
    except Exception as e:
        print(f"Unexpected error for {server_ip}: {e}")
        return None

def check_server_disk(server):
    """
    Check disk usage for a single server
    Returns status and message
    """
    print(f"\n{'='*50}")
    print(f"Checking {server['name']} ({server['ip']})...")
    
    usage = get_disk_usage(server["ip"])
    
    if usage is None:
        return "UNKNOWN", f"Could not check {server['name']}"
    
    if usage >= CRITICAL_THRESHOLD:
        status = "CRITICAL"
        message = f"⚠️ CRITICAL: {usage}% disk usage! Immediate action required!"
    elif usage >= WARNING_THRESHOLD:
        status = "WARNING"
        message = f"⚠️ WARNING: {usage}% disk usage - needs attention"
    else:
        status = "OK"
        message = f"✓ OK: {usage}% disk usage"
    
    print(message)
    return status, message, usage

def send_alert(server, status, message):
    """
    Send alert (print to console - would email/Slack in production)
    """
    alert = {
        "timestamp": datetime.now().isoformat(),
        "server": server["name"],
        "status": status,
        "message": message
    }
    print(f"\n📧 ALERT: {json.dumps(alert, indent=2)}")
    
    # In production, you'd call:
    # - send_slack_webhook()
    # - send_email()
    # - write_to_logstash()

def generate_report(results):
    """
    Generate a summary report
    """
    print(f"\n{'='*50}")
    print("DISK USAGE REPORT")
    print(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"{'='*50}")
    
    critical = 0
    warning = 0
    ok = 0
    
    for result in results:
        status = result["status"]
        if status == "CRITICAL":
            critical += 1
        elif status == "WARNING":
            warning += 1
        elif status == "OK":
            ok += 1
        
        print(f"{result['server']['name']:10} | {status:8} | {result.get('usage', 'N/A'):3}%")
    
    print(f"\nSUMMARY:")
    print(f"  ✓ Healthy: {ok}")
    print(f"  ⚠️ Warning: {warning}")
    print(f"  🚨 Critical: {critical}")
    
    if critical > 0:
        print(f"\n⚠️ ACTION REQUIRED: {critical} server(s) need immediate attention!")
    elif warning > 0:
        print(f"\nℹ️ Review needed: {warning} server(s) above threshold")
    else:
        print(f"\n✓ All servers healthy")

# Main execution
def main():
    print("Starting disk space monitor...")
    results = []
    
    for server in SERVERS:
        status, message, usage = check_server_disk(server)
        results.append({
            "server": server,
            "status": status,
            "message": message,
            "usage": usage
        })
        
        if status in ["WARNING", "CRITICAL"]:
            send_alert(server, status, message)
    
    generate_report(results)

if __name__ == "__main__":
    main()