Portal | Level: L0: Entry | Topics: Python Automation | Domain: DevOps & Tooling

Python for DevOps Drills

Gotcha: subprocess.run("kubectl get pods | grep Error", shell=True) works but is a security risk — shell injection via untrusted input. Always pass commands as a list: subprocess.run(["kubectl", "get", "pods"], ...) and pipe in Python, not in shell. The shell=True form is also slower (spawns /bin/sh).

Remember: Python's yaml.load() is unsafe — it can execute arbitrary code from YAML. Always use yaml.safe_load(). This is a common security audit finding. The unsafe version was the default in PyYAML for years, so legacy scripts often have it wrong.

One-liner: For quick JSON processing in shell scripts, python3 -c "import sys,json; d=json.load(sys.stdin); print(d['key'])" is a portable alternative to jq when jq is not installed. Every Linux box with Python 3 has json in the standard library.

Drill 1: Parse JSON API Response

Difficulty: Easy

Q: Write a Python one-liner to fetch a URL and print the JSON response pretty-printed.

Answer

# One-liner:
python3 -c "import json, urllib.request; print(json.dumps(json.loads(urllib.request.urlopen('http://localhost:8080/health').read()), indent=2))"

# In a script:
import json
import urllib.request

resp = urllib.request.urlopen('http://localhost:8080/health')
data = json.loads(resp.read())
print(json.dumps(data, indent=2))

# With requests (if installed):
import requests
r = requests.get('http://localhost:8080/health')
print(r.json())

---

Drill 2: Read and Filter YAML

Difficulty: Easy

Q: Write a Python script to read a Kubernetes YAML file and print all container image names.

Answer

import yaml

with open('deployment.yaml') as f:
    doc = yaml.safe_load(f)

containers = doc['spec']['template']['spec']['containers']
for c in containers:
    print(f"{c['name']}: {c['image']}")

# Handle multi-document YAML:
with open('manifests.yaml') as f:
    for doc in yaml.safe_load_all(f):
        if doc and doc.get('kind') == 'Deployment':
            for c in doc['spec']['template']['spec']['containers']:
                print(c['image'])

---

Drill 3: Subprocess and Shell Commands

Difficulty: Easy

Q: Run kubectl get pods -o json from Python and parse the output.

Answer

import subprocess
import json

result = subprocess.run(
    ['kubectl', 'get', 'pods', '-o', 'json'],
    capture_output=True, text=True, check=True
)
pods = json.loads(result.stdout)

for pod in pods['items']:
    name = pod['metadata']['name']
    phase = pod['status']['phase']
    print(f"{name}: {phase}")

Key points: - Use `capture_output=True` (not `shell=True` with pipes) - Use `check=True` to raise on non-zero exit - Pass commands as list, NOT string (avoids shell injection) - `text=True` gives strings instead of bytes

---

Drill 4: File Processing with pathlib

Difficulty: Easy

Q: Find all .yaml files in a directory tree and count total lines.

Answer

from pathlib import Path

total = 0
for f in Path('.').rglob('*.yaml'):
    lines = f.read_text().splitlines()
    total += len(lines)
    print(f"{f}: {len(lines)} lines")

print(f"\nTotal: {total} lines")

pathlib essentials:

p = Path('/etc/nginx/nginx.conf')
p.exists()          # True/False
p.read_text()       # Read entire file
p.write_text(data)  # Write file
p.stem              # 'nginx'
p.suffix            # '.conf'
p.parent            # Path('/etc/nginx')
p.name              # 'nginx.conf'

---

Drill 5: Environment Variables and Config

Difficulty: Easy

Q: Write a function that reads config from environment variables with defaults and validation.

Answer

import os
import sys

def get_config():
    config = {
        'db_host': os.environ.get('DB_HOST', 'localhost'),
        'db_port': int(os.environ.get('DB_PORT', '5432')),
        'db_name': os.environ['DB_NAME'],  # Required — raises KeyError
        'debug': os.environ.get('DEBUG', 'false').lower() == 'true',
        'log_level': os.environ.get('LOG_LEVEL', 'INFO'),
    }
    return config

# Safer pattern with validation:
def require_env(name):
    val = os.environ.get(name)
    if not val:
        print(f"ERROR: {name} environment variable is required", file=sys.stderr)
        sys.exit(1)
    return val

db_host = require_env('DB_HOST')

---

Drill 6: HTTP Health Check Script

Difficulty: Medium

Q: Write a script that checks health endpoints for multiple services and reports status.

Answer

import urllib.request
import json
import sys

SERVICES = {
    'api': 'http://localhost:8080/health',
    'frontend': 'http://localhost:3000/health',
    'db-proxy': 'http://localhost:6432/health',
}

def check_health(name, url, timeout=5):
    try:
        req = urllib.request.urlopen(url, timeout=timeout)
        status = req.getcode()
        return status == 200
    except Exception as e:
        return False

failures = []
for name, url in SERVICES.items():
    healthy = check_health(name, url)
    status = "OK" if healthy else "FAIL"
    print(f"  {name}: {status}")
    if not healthy:
        failures.append(name)

if failures:
    print(f"\nUnhealthy: {', '.join(failures)}")
    sys.exit(1)
else:
    print("\nAll services healthy")

---

Drill 7: Log Parsing

Difficulty: Medium

Q: Parse nginx access logs and report the top 10 IPs by request count.

Answer

from collections import Counter
import re

LOG_PATTERN = re.compile(r'^(\S+)')  # First field = IP

ip_counts = Counter()
with open('/var/log/nginx/access.log') as f:
    for line in f:
        match = LOG_PATTERN.match(line)
        if match:
            ip_counts[match.group(1)] += 1

print("Top 10 IPs by request count:")
for ip, count in ip_counts.most_common(10):
    print(f"  {ip}: {count}")

For structured JSON logs:

import json
from collections import Counter

status_counts = Counter()
with open('app.log') as f:
    for line in f:
        entry = json.loads(line)
        status_counts[entry.get('status', 'unknown')] += 1

for status, count in status_counts.most_common():
    print(f"  {status}: {count}")

---

Drill 8: Jinja2 Templating

Difficulty: Medium

Q: Generate Kubernetes manifests from a Jinja2 template with variable substitution.

Answer

from jinja2 import Template

TEMPLATE = """
apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ name }}
  namespace: {{ namespace }}
spec:
  replicas: {{ replicas }}
  selector:
    matchLabels:
      app: {{ name }}
  template:
    spec:
      containers:
      - name: {{ name }}
        image: {{ image }}:{{ tag }}
        resources:
          requests:
            cpu: "{{ cpu_request }}"
            memory: "{{ mem_request }}"
"""

services = [
    {'name': 'api', 'namespace': 'prod', 'replicas': 3,
     'image': 'myapp/api', 'tag': 'v2.1',
     'cpu_request': '250m', 'mem_request': '256Mi'},
    {'name': 'worker', 'namespace': 'prod', 'replicas': 2,
     'image': 'myapp/worker', 'tag': 'v2.1',
     'cpu_request': '500m', 'mem_request': '512Mi'},
]

tmpl = Template(TEMPLATE)
for svc in services:
    print("---")
    print(tmpl.render(**svc))

---

Drill 9: Retry Logic with Backoff

Difficulty: Medium

Q: Write a retry decorator that retries a function up to 3 times with exponential backoff.

Answer

import time
import functools

def retry(max_attempts=3, base_delay=1, backoff_factor=2):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts:
                        raise
                    delay = base_delay * (backoff_factor ** (attempt - 1))
                    print(f"Attempt {attempt} failed: {e}. Retrying in {delay}s...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, base_delay=2)
def deploy_service(name):
    # ... deployment logic that might fail
    pass

For production, use the `tenacity` library:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def call_api():
    pass

---

Drill 10: Kubernetes Client Library

Difficulty: Hard

Q: Use the Python Kubernetes client to list all pods in CrashLoopBackOff state across all namespaces.

Answer

from kubernetes import client, config

# Load kubeconfig (or in-cluster config)
try:
    config.load_incluster_config()
except config.ConfigException:
    config.load_kube_config()

v1 = client.CoreV1Api()
pods = v1.list_pod_for_all_namespaces()

crash_pods = []
for pod in pods.items:
    for cs in (pod.status.container_statuses or []):
        waiting = cs.state.waiting
        if waiting and waiting.reason == 'CrashLoopBackOff':
            crash_pods.append({
                'namespace': pod.metadata.namespace,
                'name': pod.metadata.name,
                'container': cs.name,
                'restarts': cs.restart_count,
            })

if crash_pods:
    print(f"Found {len(crash_pods)} containers in CrashLoopBackOff:")
    for p in crash_pods:
        print(f"  {p['namespace']}/{p['name']} ({p['container']}) - {p['restarts']} restarts")
else:
    print("No CrashLoopBackOff pods found")

---

Wiki Navigation

Prerequisites

• Python Exercises (Quest Ladder) (CLI) (Exercise Set, L0)

Related Content

• Perl Flashcards (CLI) (flashcard_deck, L1) — Python Automation
• Python Async & Concurrency (Topic Pack, L2) — Python Automation
• Python Debugging (Topic Pack, L1) — Python Automation
• Python Exercises (Quest Ladder) (CLI) (Exercise Set, L0) — Python Automation
• Python Flashcards (CLI) (flashcard_deck, L1) — Python Automation
• Python Packaging (Topic Pack, L2) — Python Automation
• Python for Infrastructure (Topic Pack, L1) — Python Automation
• Skillcheck: Python Automation (Assessment, L0) — Python Automation
• Software Development Flashcards (CLI) (flashcard_deck, L1) — Python Automation

Pages that link here

• Drills
• Python Async & Concurrency
• Python Async & Concurrency - Primer
• Python Debugging
• Python Debugging - Primer
• Python Packaging
• Python Packaging - Primer
• Python for DevOps Automation - Skill Check
• Python for Infrastructure
• Python for Infrastructure - Primer