Python Scripting for DevOps

Python scripting is a cornerstone of DevOps automation, enabling teams to automate repetitive tasks, orchestrate deployments, and manage infrastructure. This post covers practical Python scripting patterns and techniques specifically designed for DevOps workflows, from simple automation scripts to complex orchestration tools.

Script Structure Best Practices

Shebang and Executable Scripts

#!/usr/bin/env python3
"""
Server deployment automation script.

This script automates the deployment of applications to servers.
"""

import sys
import argparse
from pathlib import Path

def main():
    """Main entry point for the script."""
    parser = argparse.ArgumentParser(description="Deploy application to servers")
    parser.add_argument("--environment", required=True, choices=["dev", "staging", "prod"])
    parser.add_argument("--version", required=True, help="Version to deploy")
    parser.add_argument("--dry-run", action="store_true", help="Simulate deployment")
    
    args = parser.parse_args()
    
    print(f"Deploying version {args.version} to {args.environment}")
    
    if args.dry_run:
        print("DRY RUN - No actual changes made")
        return 0
    
    # Deployment logic here
    return 0

if __name__ == "__main__":
    sys.exit(main())

Make the script executable:

chmod +x deploy.py
./deploy.py --environment dev --version 1.2.3

Configuration Management

import os
import json
import yaml
from pathlib import Path
from typing import Dict, Any

class Config:
    """Configuration manager for scripts."""
    
    def __init__(self, config_file: str = None):
        self.config = {}
        
        # Load from environment
        self.load_from_env()
        
        # Load from file
        if config_file:
            self.load_from_file(config_file)
    
    def load_from_env(self):
        """Load configuration from environment variables."""
        self.config["api_url"] = os.getenv("API_URL", "http://localhost:8080")
        self.config["timeout"] = int(os.getenv("TIMEOUT", "30"))
        self.config["debug"] = os.getenv("DEBUG", "false").lower() == "true"
    
    def load_from_file(self, filepath: str):
        """Load configuration from YAML or JSON file."""
        path = Path(filepath)
        
        if not path.exists():
            raise FileNotFoundError(f"Config file not found: {filepath}")
        
        with open(path, "r") as f:
            if path.suffix == ".yaml" or path.suffix == ".yml":
                file_config = yaml.safe_load(f)
            elif path.suffix == ".json":
                file_config = json.load(f)
            else:
                raise ValueError(f"Unsupported config format: {path.suffix}")
        
        # Merge with existing config (file takes precedence)
        self.config.update(file_config)
    
    def get(self, key: str, default: Any = None) -> Any:
        """Get configuration value."""
        return self.config.get(key, default)

# Usage
config = Config("config.yaml")
api_url = config.get("api_url")

graph LR
    A[Script Start] --> B[Load Environment Variables]
    B --> C[Load Config File]
    C --> D[Merge Configuration]
    D --> E[Use Config in Script]

Running System Commands

import subprocess
from typing import Tuple

def run_command(command: list, capture_output: bool = True) -> Tuple[int, str, str]:
    """
    Run a system command and return exit code, stdout, stderr.
    
    Args:
        command: Command as list of strings
        capture_output: Whether to capture stdout/stderr
    
    Returns:
        Tuple of (exit_code, stdout, stderr)
    """
    try:
        result = subprocess.run(
            command,
            capture_output=capture_output,
            text=True,
            check=False
        )
        return result.returncode, result.stdout, result.stderr
    except Exception as e:
        return 1, "", str(e)

# Usage
exit_code, stdout, stderr = run_command(["docker", "ps", "-a"])

if exit_code == 0:
    print("Command succeeded:")
    print(stdout)
else:
    print(f"Command failed: {stderr}")

# Run command with shell
def run_shell_command(command: str) -> Tuple[int, str, str]:
    """Run command in shell (use with caution)."""
    result = subprocess.run(
        command,
        shell=True,
        capture_output=True,
        text=True
    )
    return result.returncode, result.stdout, result.stderr

# Execute multiple commands in pipeline
def run_pipeline(commands: list) -> str:
    """Execute commands in a pipeline."""
    processes = []
    
    for i, cmd in enumerate(commands):
        stdin = processes[i-1].stdout if i > 0 else None
        proc = subprocess.Popen(
            cmd,
            stdin=stdin,
            stdout=subprocess.PIPE,
            text=True
        )
        processes.append(proc)
    
    output, _ = processes[-1].communicate()
    return output

# Usage: docker ps | grep nginx
output = run_pipeline([["docker", "ps"], ["grep", "nginx"]])

Logging and Error Handling

import logging
import sys
from datetime import datetime

def setup_logging(log_file: str = None, level: str = "INFO"):
    """Configure logging for the script."""
    log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    
    handlers = [logging.StreamHandler(sys.stdout)]
    
    if log_file:
        handlers.append(logging.FileHandler(log_file))
    
    logging.basicConfig(
        level=getattr(logging, level.upper()),
        format=log_format,
        handlers=handlers
    )

# Usage
setup_logging(log_file="deployment.log", level="INFO")
logger = logging.getLogger(__name__)

logger.info("Starting deployment")
logger.warning("Configuration file not found, using defaults")
logger.error("Failed to connect to server")

# Context manager for operation logging
from contextlib import contextmanager
import time

@contextmanager
def log_operation(operation_name: str):
    """Context manager to log operation start, end, and duration."""
    logger.info(f"Starting: {operation_name}")
    start_time = time.time()
    
    try:
        yield
        duration = time.time() - start_time
        logger.info(f"Completed: {operation_name} (took {duration:.2f}s)")
    except Exception as e:
        duration = time.time() - start_time
        logger.error(f"Failed: {operation_name} after {duration:.2f}s - {e}")
        raise

# Usage
with log_operation("Database backup"):
    backup_database()

File and Directory Operations

import shutil
from pathlib import Path
from typing import List

def ensure_directory(path: str):
    """Create directory if it doesn't exist."""
    Path(path).mkdir(parents=True, exist_ok=True)

def clean_directory(path: str, pattern: str = "*"):
    """Remove all files matching pattern in directory."""
    directory = Path(path)
    for item in directory.glob(pattern):
        if item.is_file():
            item.unlink()
        elif item.is_dir():
            shutil.rmtree(item)

def copy_files(source: str, dest: str, pattern: str = "*"):
    """Copy files matching pattern from source to dest."""
    source_path = Path(source)
    dest_path = Path(dest)
    
    ensure_directory(dest)
    
    for file_path in source_path.glob(pattern):
        if file_path.is_file():
            shutil.copy2(file_path, dest_path / file_path.name)

def find_files(directory: str, pattern: str, recursive: bool = True) -> List[Path]:
    """Find all files matching pattern."""
    path = Path(directory)
    glob_pattern = f"**/{pattern}" if recursive else pattern
    return list(path.glob(glob_pattern))

# Archive operations
def create_tar_archive(source_dir: str, output_file: str):
    """Create tar.gz archive."""
    import tarfile
    
    with tarfile.open(output_file, "w:gz") as tar:
        tar.add(source_dir, arcname=Path(source_dir).name)

def extract_tar_archive(archive_file: str, dest_dir: str):
    """Extract tar.gz archive."""
    import tarfile
    
    with tarfile.open(archive_file, "r:gz") as tar:
        tar.extractall(dest_dir)

# Usage
ensure_directory("backups")
create_tar_archive("/var/www/app", "backups/app-backup.tar.gz")

Working with Remote Servers

import paramiko
from typing import Optional

class SSHClient:
    """Simplified SSH client for remote operations."""
    
    def __init__(self, hostname: str, username: str, 
                 password: Optional[str] = None, 
                 key_filename: Optional[str] = None):
        self.client = paramiko.SSHClient()
        self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        if key_filename:
            self.client.connect(
                hostname, 
                username=username, 
                key_filename=key_filename
            )
        else:
            self.client.connect(
                hostname, 
                username=username, 
                password=password
            )
    
    def execute(self, command: str) -> Tuple[int, str, str]:
        """Execute command on remote server."""
        stdin, stdout, stderr = self.client.exec_command(command)
        exit_code = stdout.channel.recv_exit_status()
        return exit_code, stdout.read().decode(), stderr.read().decode()
    
    def upload_file(self, local_path: str, remote_path: str):
        """Upload file to remote server."""
        sftp = self.client.open_sftp()
        sftp.put(local_path, remote_path)
        sftp.close()
    
    def download_file(self, remote_path: str, local_path: str):
        """Download file from remote server."""
        sftp = self.client.open_sftp()
        sftp.get(remote_path, local_path)
        sftp.close()
    
    def close(self):
        """Close SSH connection."""
        self.client.close()

# Usage
with SSHClient("server.example.com", "deploy", key_filename="~/.ssh/id_rsa") as ssh:
    exit_code, stdout, stderr = ssh.execute("docker ps")
    print(stdout)

API Integration

import requests
from typing import Dict, Any, Optional
import time

class APIClient:
    """Reusable API client with retry logic."""
    
    def __init__(self, base_url: str, api_key: Optional[str] = None, 
                 timeout: int = 30, max_retries: int = 3):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.max_retries = max_retries
        self.session = requests.Session()
        
        if api_key:
            self.session.headers.update({"Authorization": f"Bearer {api_key}"})
    
    def _request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        """Make HTTP request with retry logic."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        for attempt in range(self.max_retries):
            try:
                response = self.session.request(
                    method, 
                    url, 
                    timeout=self.timeout, 
                    **kwargs
                )
                response.raise_for_status()
                return response
            except requests.exceptions.RequestException as e:
                if attempt == self.max_retries - 1:
                    raise
                
                wait_time = 2 ** attempt
                logger.warning(f"Request failed, retrying in {wait_time}s: {e}")
                time.sleep(wait_time)
    
    def get(self, endpoint: str, **kwargs) -> Dict[Any, Any]:
        """GET request."""
        response = self._request("GET", endpoint, **kwargs)
        return response.json()
    
    def post(self, endpoint: str, data: Dict = None, **kwargs) -> Dict[Any, Any]:
        """POST request."""
        response = self._request("POST", endpoint, json=data, **kwargs)
        return response.json()
    
    def put(self, endpoint: str, data: Dict = None, **kwargs) -> Dict[Any, Any]:
        """PUT request."""
        response = self._request("PUT", endpoint, json=data, **kwargs)
        return response.json()
    
    def delete(self, endpoint: str, **kwargs) -> Dict[Any, Any]:
        """DELETE request."""
        response = self._request("DELETE", endpoint, **kwargs)
        return response.json() if response.text else {}

# Usage
api = APIClient("https://api.example.com", api_key="secret-key")
users = api.get("/users")
user = api.post("/users", data={"name": "Alice", "email": "alice@example.com"})

sequenceDiagram
    participant Script
    participant APIClient
    participant API
    
    Script->>APIClient: get("/users")
    APIClient->>API: HTTP GET
    alt Success
        API->>APIClient: 200 OK
        APIClient->>Script: Return data
    else Failure
        API->>APIClient: Error
        APIClient->>APIClient: Retry (exponential backoff)
        APIClient->>API: HTTP GET (retry)
    end

Docker Operations

import docker
from typing import List, Dict

class DockerManager:
    """Manage Docker containers and images."""
    
    def __init__(self):
        self.client = docker.from_env()
    
    def list_containers(self, all: bool = False) -> List[Dict]:
        """List running containers."""
        containers = self.client.containers.list(all=all)
        return [
            {
                "id": c.short_id,
                "name": c.name,
                "status": c.status,
                "image": c.image.tags[0] if c.image.tags else "none"
            }
            for c in containers
        ]
    
    def run_container(self, image: str, name: str = None, 
                     ports: Dict = None, environment: Dict = None) -> str:
        """Run a new container."""
        container = self.client.containers.run(
            image,
            name=name,
            ports=ports,
            environment=environment,
            detach=True
        )
        return container.id
    
    def stop_container(self, container_id: str):
        """Stop a running container."""
        container = self.client.containers.get(container_id)
        container.stop()
    
    def remove_container(self, container_id: str, force: bool = False):
        """Remove a container."""
        container = self.client.containers.get(container_id)
        container.remove(force=force)
    
    def build_image(self, path: str, tag: str) -> bool:
        """Build Docker image from Dockerfile."""
        try:
            self.client.images.build(path=path, tag=tag)
            return True
        except docker.errors.BuildError as e:
            logger.error(f"Build failed: {e}")
            return False
    
    def pull_image(self, image: str, tag: str = "latest"):
        """Pull image from registry."""
        self.client.images.pull(image, tag=tag)
    
    def cleanup_dangling_images(self):
        """Remove dangling images."""
        self.client.images.prune(filters={"dangling": True})

# Usage
docker_mgr = DockerManager()
containers = docker_mgr.list_containers(all=True)
for container in containers:
    print(f"{container['name']}: {container['status']}")

Practical Example: Deployment Automation Script

#!/usr/bin/env python3
"""
Automated application deployment script.

Handles building, testing, and deploying applications to various environments.
"""

import sys
import argparse
import logging
from pathlib import Path
from typing import Dict
import subprocess
import time

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

class DeploymentManager:
    """Manage application deployment process."""
    
    def __init__(self, environment: str, version: str, dry_run: bool = False):
        self.environment = environment
        self.version = version
        self.dry_run = dry_run
        self.config = self.load_config()
    
    def load_config(self) -> Dict:
        """Load environment-specific configuration."""
        import yaml
        
        config_file = f"config/{self.environment}.yaml"
        with open(config_file, "r") as f:
            return yaml.safe_load(f)
    
    def run_command(self, command: list) -> bool:
        """Execute command and return success status."""
        logger.info(f"Running: {' '.join(command)}")
        
        if self.dry_run:
            logger.info("DRY RUN - Command not executed")
            return True
        
        result = subprocess.run(command, capture_output=True, text=True)
        
        if result.returncode != 0:
            logger.error(f"Command failed: {result.stderr}")
            return False
        
        logger.info(result.stdout)
        return True
    
    def build(self) -> bool:
        """Build application."""
        logger.info(f"Building version {self.version}")
        
        commands = [
            ["docker", "build", "-t", f"myapp:{self.version}", "."],
            ["docker", "tag", f"myapp:{self.version}", 
             f"registry.example.com/myapp:{self.version}"]
        ]
        
        for cmd in commands:
            if not self.run_command(cmd):
                return False
        
        return True
    
    def test(self) -> bool:
        """Run tests."""
        logger.info("Running tests")
        
        command = [
            "docker", "run", "--rm", f"myapp:{self.version}",
            "pytest", "tests/"
        ]
        
        return self.run_command(command)
    
    def push(self) -> bool:
        """Push image to registry."""
        logger.info("Pushing image to registry")
        
        command = [
            "docker", "push", 
            f"registry.example.com/myapp:{self.version}"
        ]
        
        return self.run_command(command)
    
    def deploy(self) -> bool:
        """Deploy to environment."""
        logger.info(f"Deploying to {self.environment}")
        
        # Update Kubernetes deployment
        commands = [
            ["kubectl", "set", "image", "deployment/myapp",
             f"myapp=registry.example.com/myapp:{self.version}",
             f"--namespace={self.environment}"],
            ["kubectl", "rollout", "status", "deployment/myapp",
             f"--namespace={self.environment}"]
        ]
        
        for cmd in commands:
            if not self.run_command(cmd):
                return False
        
        return True
    
    def rollback(self) -> bool:
        """Rollback deployment."""
        logger.warning("Rolling back deployment")
        
        command = [
            "kubectl", "rollout", "undo", "deployment/myapp",
            f"--namespace={self.environment}"
        ]
        
        return self.run_command(command)
    
    def execute(self) -> int:
        """Execute full deployment process."""
        try:
            logger.info(f"Starting deployment to {self.environment}")
            logger.info(f"Version: {self.version}")
            logger.info(f"Dry run: {self.dry_run}")
            
            if not self.build():
                logger.error("Build failed")
                return 1
            
            if not self.test():
                logger.error("Tests failed")
                return 1
            
            if not self.push():
                logger.error("Push failed")
                return 1
            
            if not self.deploy():
                logger.error("Deployment failed, initiating rollback")
                self.rollback()
                return 1
            
            logger.info("Deployment completed successfully")
            return 0
            
        except Exception as e:
            logger.error(f"Deployment error: {e}")
            return 1

def main():
    parser = argparse.ArgumentParser(description="Deploy application")
    parser.add_argument(
        "--environment", 
        required=True,
        choices=["dev", "staging", "prod"],
        help="Target environment"
    )
    parser.add_argument(
        "--version",
        required=True,
        help="Version to deploy"
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Simulate deployment without making changes"
    )
    
    args = parser.parse_args()
    
    manager = DeploymentManager(
        environment=args.environment,
        version=args.version,
        dry_run=args.dry_run
    )
    
    return manager.execute()

if __name__ == "__main__":
    sys.exit(main())

Key Takeaways