Python for DevOps Master Class #

Engineering-Grade Reference Manual for Python in DevOps
Automation, cloud SDKs, and DevOps tooling with Python

---

Table of Contents #

1. Python Basics for DevOps
2. File & System Operations
3. AWS SDK (boto3)
4. Azure SDK
5. Kubernetes Client
6. API Automation
7. CLI Tools with Click
8. Testing & Quality
9. Common DevOps Scripts

---

Python Basics for DevOps #

🔹 Essential Libraries #

import os          # OS operations
import sys         # System operations
import subprocess  # Run commands
import json        # JSON handling
import yaml        # YAML handling
import requests    # HTTP requests
import logging     # Logging
from pathlib import Path  # File paths

🔹 Running Commands #

import subprocess

# Run command
result = subprocess.run(['ls', '-la'], capture_output=True, text=True)
print(result.stdout)

# Check if command succeeded
if result.returncode == 0:
    print("Success")
else:
    print(f"Error: {result.stderr}")

# Run with shell
subprocess.run('echo $HOME', shell=True)

---

File & System Operations #

🔹 File Operations #

from pathlib import Path

# Read file
content = Path('file.txt').read_text()

# Write file
Path('output.txt').write_text('Hello World')

# Check if file exists
if Path('file.txt').exists():
    print("File exists")

# List files
for file in Path('.').glob('*.txt'):
    print(file)

# Create directory
Path('/tmp/mydir').mkdir(parents=True, exist_ok=True)

---

AWS SDK (boto3) #

🔹 EC2 Operations #

import boto3

ec2 = boto3.client('ec2')

# List instances
response = ec2.describe_instances()
for reservation in response['Reservations']:
    for instance in reservation['Instances']:
        print(f"Instance ID: {instance['InstanceId']}")
        print(f"State: {instance['State']['Name']}")

# Start instance
ec2.start_instances(InstanceIds=['i-1234567890abcdef0'])

# Stop instance
ec2.stop_instances(InstanceIds=['i-1234567890abcdef0'])

🔹 S3 Operations #

s3 = boto3.client('s3')

# List buckets
response = s3.list_buckets()
for bucket in response['Buckets']:
    print(bucket['Name'])

# Upload file
s3.upload_file('local.txt', 'my-bucket', 'remote.txt')

# Download file
s3.download_file('my-bucket', 'remote.txt', 'local.txt')

# List objects
response = s3.list_objects_v2(Bucket='my-bucket')
for obj in response.get('Contents', []):
    print(obj['Key'])

---

Azure SDK #

🔹 Virtual Machines #

from azure.identity import DefaultAzureCredential
from azure.mgmt.compute import ComputeManagementClient

credential = DefaultAzureCredential()
compute_client = ComputeManagementClient(credential, subscription_id)

# List VMs
for vm in compute_client.virtual_machines.list_all():
    print(f"VM: {vm.name}")

# Start VM
compute_client.virtual_machines.begin_start(
    resource_group_name,
    vm_name
)

---

Kubernetes Client #

🔹 Pod Operations #

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

# List pods
pods = v1.list_namespaced_pod('default')
for pod in pods.items:
    print(f"Pod: {pod.metadata.name}")
    print(f"Status: {pod.status.phase}")

# Create pod
pod_manifest = {
    'apiVersion': 'v1',
    'kind': 'Pod',
    'metadata': {'name': 'test-pod'},
    'spec': {
        'containers': [{
            'name': 'nginx',
            'image': 'nginx:latest'
        }]
    }
}
v1.create_namespaced_pod('default', pod_manifest)

# Delete pod
v1.delete_namespaced_pod('test-pod', 'default')

---

API Automation #

🔹 REST API Calls #

import requests

# GET request
response = requests.get('https://api.example.com/users')
users = response.json()

# POST request
data = {'name': 'John', 'email': 'john@example.com'}
response = requests.post('https://api.example.com/users', json=data)

# With authentication
headers = {'Authorization': 'Bearer token123'}
response = requests.get('https://api.example.com/data', headers=headers)

# Error handling
try:
    response = requests.get('https://api.example.com/data')
    response.raise_for_status()
    data = response.json()
except requests.exceptions.HTTPError as e:
    print(f"HTTP Error: {e}")
except requests.exceptions.RequestException as e:
    print(f"Error: {e}")

---

CLI Tools with Click #

🔹 Building CLI Tools #

import click

@click.group()
def cli():
    """DevOps CLI Tool"""
    pass

@cli.command()
@click.option('--name', prompt='Your name', help='The person to greet')
def hello(name):
    """Say hello"""
    click.echo(f'Hello {name}!')

@cli.command()
@click.argument('filename')
@click.option('--lines', default=10, help='Number of lines')
def tail(filename, lines):
    """Show last N lines of file"""
    with open(filename) as f:
        all_lines = f.readlines()
        for line in all_lines[-lines:]:
            click.echo(line, nl=False)

if __name__ == '__main__':
    cli()

---

Testing & Quality #

🔹 Unit Testing #

import pytest

def add(a, b):
    return a + b

def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0

# Run with: pytest test_file.py

---

Common DevOps Scripts #

🔹 Log Parser #

import re
from collections import Counter

def parse_nginx_logs(log_file):
    ip_pattern = r'(\d+\.\d+\.\d+\.\d+)'
    ips = []
    
    with open(log_file) as f:
        for line in f:
            match = re.search(ip_pattern, line)
            if match:
                ips.append(match.group(1))
    
    # Count occurrences
    ip_counts = Counter(ips)
    
    # Top 10 IPs
    for ip, count in ip_counts.most_common(10):
        print(f"{ip}: {count} requests")

parse_nginx_logs('/var/log/nginx/access.log')

🔹 Health Check Script #

import requests
import smtplib
from email.mime.text import MIMEText

def check_service(url):
    try:
        response = requests.get(url, timeout=5)
        return response.status_code == 200
    except:
        return False

def send_alert(service_name):
    msg = MIMEText(f'{service_name} is down!')
    msg['Subject'] = f'Alert: {service_name} Down'
    msg['From'] = 'alerts@example.com'
    msg['To'] = 'admin@example.com'
    
    with smtplib.SMTP('localhost') as server:
        server.send_message(msg)

services = {
    'Website': 'https://example.com',
    'API': 'https://api.example.com/health'
}

for name, url in services.items():
    if not check_service(url):
        print(f'{name} is down!')
        send_alert(name)

---

🎓 Python for DevOps Master Class Complete