#Python for DevOps

Python is the most popular language in the DevOps ecosystem, powering tools like Ansible, SaltStack, and countless automation scripts.

#📋 Table of Contents

Why Python for DevOps?
Setting Up Your Environment
Python Basics
File Operations
Working with APIs
Automation Scripts
Popular DevOps Libraries

#Why Python for DevOps?

#Key Advantages

Advantage	Description
Readability	Clean syntax that's easy to learn and maintain
Ecosystem	Rich library ecosystem for every DevOps need
Cross-Platform	Works on Linux, Windows, and macOS
Community	Massive community with extensive documentation
Integration	Easy integration with cloud providers and tools

#Real-World Use Cases

Loading diagram...

#Setting Up Your Environment

#Installation

bash

1# Ubuntu/Debian
2sudo apt update
3sudo apt install python3 python3-pip python3-venv
4
5# CentOS/RHEL
6sudo yum install python3 python3-pip
7
8# macOS (using Homebrew)
9brew install python3
10
11# Windows (using Chocolatey)
12choco install python3

#Virtual Environments

Always use virtual environments to isolate project dependencies:

bash

1# Create a virtual environment
2python3 -m venv devops-env
3
4# Activate it (Linux/macOS)
5source devops-env/bin/activate
6
7# Activate it (Windows)
8devops-env\Scripts\activate
9
10# Install packages
11pip install boto3 requests paramiko pyyaml

#Essential DevOps Packages

bash

1pip install \
2    boto3 \          # AWS SDK
3    azure-mgmt \     # Azure SDK
4    google-cloud \   # GCP SDK
5    paramiko \       # SSH library
6    requests \       # HTTP library
7    pyyaml \         # YAML parsing
8    jinja2 \         # Templating
9    click \          # CLI framework
10    fabric \         # Remote execution
11    netmiko          # Network automation

#Python Basics

#Variables and Data Types

python

1# Strings
2server_name = "web-server-01"
3region = "us-east-1"
4
5# Numbers
6port = 8080
7cpu_threshold = 80.5
8
9# Lists
10servers = ["web-01", "web-02", "web-03"]
11ports = [80, 443, 8080]
12
13# Dictionaries
14server_config = {
15    "name": "web-server-01",
16    "ip": "10.0.1.100",
17    "port": 8080,
18    "enabled": True
19}
20
21# f-strings (formatted strings)
22print(f"Server {server_name} is running on port {port}")

#Control Flow

python

1# If statements
2cpu_usage = 85
3
4if cpu_usage > 90:
5    print("CRITICAL: CPU usage is very high!")
6elif cpu_usage > 75:
7    print("WARNING: CPU usage is elevated")
8else:
9    print("OK: CPU usage is normal")
10
11# For loops
12servers = ["web-01", "web-02", "web-03"]
13for server in servers:
14    print(f"Checking server: {server}")
15
16# List comprehension
17healthy_servers = [s for s in servers if check_health(s)]

#Functions

python

1def check_server_health(hostname: str, port: int = 80) -> bool:
2    """
3    Check if a server is responding on the specified port.
4    
5    Args:
6        hostname: Server hostname or IP
7        port: Port to check (default: 80)
8    
9    Returns:
10        True if server is healthy, False otherwise
11    """
12    import socket
13    
14    try:
15        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
16        sock.settimeout(5)
17        result = sock.connect_ex((hostname, port))
18        sock.close()
19        return result == 0
20    except Exception as e:
21        print(f"Error checking {hostname}: {e}")
22        return False
23
24# Usage
25if check_server_health("google.com", 443):
26    print("Server is healthy!")

#Classes

python

1class Server:
2    """Represents a server in our infrastructure."""
3    
4    def __init__(self, name: str, ip: str, environment: str = "production"):
5        self.name = name
6        self.ip = ip
7        self.environment = environment
8        self.is_healthy = True
9    
10    def check_health(self) -> bool:
11        """Perform health check on the server."""
12        # Implementation here
13        return self.is_healthy
14    
15    def restart(self) -> None:
16        """Restart the server."""
17        print(f"Restarting {self.name}...")
18        # Implementation here
19    
20    def __repr__(self) -> str:
21        return f"Server(name={self.name}, ip={self.ip})"
22
23# Usage
24web_server = Server("web-01", "10.0.1.100")
25print(web_server.check_health())

#File Operations

#Reading and Writing Files

python

1# Reading a file
2with open("/etc/hosts", "r") as f:
3    content = f.read()
4    print(content)
5
6# Reading line by line
7with open("/var/log/syslog", "r") as f:
8    for line in f:
9        if "error" in line.lower():
10            print(line.strip())
11
12# Writing to a file
13config = """
14server {
15    listen 80;
16    server_name example.com;
17}
18"""
19
20with open("/tmp/nginx.conf", "w") as f:
21    f.write(config)
22
23# Appending to a file
24with open("/tmp/log.txt", "a") as f:
25    f.write("New log entry\n")

#Working with YAML

python

1import yaml
2
3# Reading YAML
4with open("config.yaml", "r") as f:
5    config = yaml.safe_load(f)
6
7print(config["database"]["host"])
8
9# Writing YAML
10infrastructure = {
11    "servers": [
12        {"name": "web-01", "ip": "10.0.1.100"},
13        {"name": "web-02", "ip": "10.0.1.101"}
14    ],
15    "database": {
16        "host": "db.example.com",
17        "port": 5432
18    }
19}
20
21with open("infrastructure.yaml", "w") as f:
22    yaml.dump(infrastructure, f, default_flow_style=False)

#Working with JSON

python

1import json
2
3# Reading JSON
4with open("config.json", "r") as f:
5    config = json.load(f)
6
7# Writing JSON
8data = {"servers": ["web-01", "web-02"], "count": 2}
9with open("output.json", "w") as f:
10    json.dump(data, f, indent=2)
11
12# Pretty printing
13print(json.dumps(data, indent=2))

#Working with APIs

#HTTP Requests with requests

python

1import requests
2
3# GET request
4response = requests.get("https://api.github.com/users/octocat")
5if response.status_code == 200:
6    user = response.json()
7    print(f"Name: {user['name']}")
8    print(f"Public repos: {user['public_repos']}")
9
10# POST request with authentication
11api_key = "your-api-key"
12headers = {
13    "Authorization": f"Bearer {api_key}",
14    "Content-Type": "application/json"
15}
16
17payload = {
18    "name": "new-server",
19    "region": "us-east-1"
20}
21
22response = requests.post(
23    "https://api.example.com/servers",
24    headers=headers,
25    json=payload
26)
27
28if response.status_code == 201:
29    print("Server created successfully!")

#AWS SDK (Boto3)

python

1import boto3
2
3# Initialize clients
4ec2 = boto3.client('ec2', region_name='us-east-1')
5s3 = boto3.client('s3')
6
7# List EC2 instances
8response = ec2.describe_instances()
9for reservation in response['Reservations']:
10    for instance in reservation['Instances']:
11        print(f"Instance ID: {instance['InstanceId']}")
12        print(f"State: {instance['State']['Name']}")
13        print(f"Type: {instance['InstanceType']}")
14        print("---")
15
16# List S3 buckets
17response = s3.list_buckets()
18for bucket in response['Buckets']:
19    print(f"Bucket: {bucket['Name']}")
20
21# Create an EC2 instance
22response = ec2.run_instances(
23    ImageId='ami-0c55b159cbfafe1f0',
24    InstanceType='t2.micro',
25    MinCount=1,
26    MaxCount=1,
27    TagSpecifications=[{
28        'ResourceType': 'instance',
29        'Tags': [{'Key': 'Name', 'Value': 'DevOps-Server'}]
30    }]
31)

#Automation Scripts

#Server Health Check Script

python

1#!/usr/bin/env python3
2"""
3Server Health Check Script
4Checks multiple servers and reports their status.
5"""
6
7import socket
8import concurrent.futures
9from datetime import datetime
10
11def check_port(host: str, port: int, timeout: int = 5) -> bool:
12    """Check if a port is open on a host."""
13    try:
14        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
15        sock.settimeout(timeout)
16        result = sock.connect_ex((host, port))
17        sock.close()
18        return result == 0
19    except Exception:
20        return False
21
22def check_server(server: dict) -> dict:
23    """Check health of a server."""
24    result = {
25        "name": server["name"],
26        "host": server["host"],
27        "port": server["port"],
28        "status": "healthy" if check_port(server["host"], server["port"]) else "unhealthy",
29        "checked_at": datetime.now().isoformat()
30    }
31    return result
32
33def main():
34    servers = [
35        {"name": "Web Server", "host": "google.com", "port": 443},
36        {"name": "API Server", "host": "api.github.com", "port": 443},
37        {"name": "Database", "host": "localhost", "port": 5432},
38    ]
39    
40    print("🔍 Starting health checks...\n")
41    
42    # Check servers in parallel
43    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
44        results = list(executor.map(check_server, servers))
45    
46    # Print results
47    for result in results:
48        status_icon = "✅" if result["status"] == "healthy" else "❌"
49        print(f"{status_icon} {result['name']}: {result['status']}")
50    
51    # Summary
52    healthy = sum(1 for r in results if r["status"] == "healthy")
53    print(f"\n📊 Summary: {healthy}/{len(results)} servers healthy")
54
55if __name__ == "__main__":
56    main()

#Log Parser Script

diagram

#!/usr/bin/env python3
"""
Log Parser Script
Parses log files and extracts useful information.
"""

import re
from collections import Counter
from datetime import datetime

def parse_apache_log(log_file: str) -> dict:
    """Parse Apache access log and return statistics."""
    
    # Apache Combined Log Format regex
    pattern = r'(\S+) \S+ \S+ \[(.*?)\] "(.*?)" (\d+) (\d+|-)'
    
    stats = {
        "total_requests": 0,
        "status_codes": Counter(),
        "ip_addresses": Counter(),
        "requests_by_hour": Counter(),
    }
    
    with open(log_file, "r") as f:
        for line in f:
            match = re.match(pattern, line)
            if match:
                ip, timestamp, request, status, size = match.groups()
                
                stats["total_requests"] += 1
                stats["status_codes"][status] += 1
                stats["ip_addresses"][ip] += 1
                
                # Extract hour
                try:
                    dt = datetime.strptime(timestamp.split()[0], "%d/%b/%Y:%H:%M:%S")
                    stats["requests_by_hour"][dt.hour] += 1
                except ValueError:
                    pass
    
    return stats

def main():
    log_file = "/var/log/apache2/access.log"
    
    try:
        stats = parse_apache_log(log_file)
        
        print("📊 Log Analysis Report")
        print("=" * 40)
        print(f"Total Requests: {stats['total_requests']}")
        print("\n📈 Status Codes:")
        for code, count in stats['status_codes'].most_common():
            print(f"  {code}: {count}")
        
        print("\n🌐 Top 5 IP Addresses:")
        for ip, count in stats['ip_addresses'].most_common(5):
            print(f"  {ip}: {count} requests")
            
    except FileNotFoundError:
        print(f"Error: Log file not found: {log_file}")

if __name__ == "__main__":
    main()

#Popular DevOps Libraries

Library	Purpose	Install
boto3	AWS SDK	`pip install boto3`
paramiko	SSH connections	`pip install paramiko`
fabric	Remote execution	`pip install fabric`
ansible	Configuration management	`pip install ansible`
requests	HTTP client	`pip install requests`
click	CLI framework	`pip install click`
pyyaml	YAML parsing	`pip install pyyaml`
jinja2	Templating	`pip install jinja2`
netmiko	Network devices	`pip install netmiko`
docker	Docker SDK	`pip install docker`
kubernetes	Kubernetes client	`pip install kubernetes`

#Best Practices

#1. Use Virtual Environments

bash

1# Always isolate your projects
2python3 -m venv venv
3source venv/bin/activate
4pip install -r requirements.txt

#2. Use Type Hints

python

def deploy_server(name: str, region: str, count: int = 1) -> list[str]:
    """Type hints make code self-documenting."""
    pass

#3. Handle Errors Gracefully

python

1try:
2    response = requests.get(url, timeout=10)
3    response.raise_for_status()
4except requests.exceptions.Timeout:
5    logger.error("Request timed out")
6except requests.exceptions.HTTPError as e:
7    logger.error(f"HTTP error: {e}")
8except Exception as e:
9    logger.error(f"Unexpected error: {e}")

#4. Use Logging Instead of Print

python

1import logging
2
3logging.basicConfig(
4    level=logging.INFO,
5    format='%(asctime)s - %(levelname)s - %(message)s'
6)
7logger = logging.getLogger(__name__)
8
9logger.info("Starting deployment...")
10logger.warning("Server is running low on memory")
11logger.error("Deployment failed!")

#Summary

Concept	Key Points
Setup	Use venv, pip for packages
Basics	Variables, functions, classes
Files	YAML, JSON, text manipulation
APIs	requests, boto3, SDKs
Automation	Health checks, log parsing, deployments

#Next Steps

Practice with the hands-on labs
Build a simple automation script for your workflow
Explore cloud SDKs (boto3, azure-sdk, google-cloud)
Learn about testing with pytest

[!TIP] Pro Tip: Start by automating something you do repeatedly. The best way to learn is by solving real problems!