Python Libraries for DevOps

Implementation Examples

This post provides practical, real-world examples for common Python libraries used in DevOps and automation. Each example demonstrates a fundamental use case for the respective library.

1. boto3       (AWS SDK for Python)
2. paramiko (SSH Protocol)
3. docker     (Docker Engine API)
4. kubernetes (Kubernetes API)
5. pyyaml        (YAML Parser)
6. requests    (HTTP Library)
7. fabric        (High-level SSH Task Execution)
8. pytest        (Testing Framework)
9. ansible-runner (Ansible from Python)
10. sys            (System-specific parameters)
11. subprocess (Running External Commands)
12. os               (Operating System Interface)
13. json           (JSON Encoder and Decoder)
14. logging      (Logging Facility)

========== Implementation .... ==========

1. boto3 (AWS SDK for Python)

Use Case: Listing all Amazon S3 buckets in your AWS account. This is a fundamental task for auditing or managing cloud storage resources.

import boto3

# Ensure your AWS credentials are configured (e.g., via ~/.aws/credentials)
s3_client = boto3.client('s3')

try:
    response = s3_client.list_buckets()
    print("Existing S3 Buckets:")
    for bucket in response['Buckets']:
        print(f'  - {bucket["Name"]}')
except Exception as e:
    print(f"An error occurred: {e}")

2. paramiko (SSH Protocol)

Use Case: Connecting to a remote server via SSH and executing a command (uptime) to check its status.

import paramiko

# --- Connection Details ---
HOSTNAME = "your_server_ip_or_hostname"
PORT = 22
USERNAME = "your_username"
PASSWORD = "your_password" # For production, always use SSH keys instead!

client = paramiko.SSHClient()
# Automatically add the server's host key (less secure, fine for demos)
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:
    client.connect(hostname=HOSTNAME, port=PORT, username=USERNAME, password=PASSWORD)
    stdin, stdout, stderr = client.exec_command('uptime')
    print("Server Uptime:")
    print(stdout.read().decode())
finally:
    client.close()

3. docker (Docker Engine API)

Use Case: Checking if the nginx:latest image exists locally, pulling it if it doesn't, and then running a new container from it.

import docker

# Connect to the Docker daemon
client = docker.from_env()

image_name = "nginx:latest"

try:
    # Check if the image exists
    client.images.get(image_name)
    print(f"Image '{image_name}' already exists locally.")
except docker.errors.ImageNotFound:
    print(f"Image '{image_name}' not found. Pulling from Docker Hub...")
    client.images.pull(image_name)
    print("Pull complete.")

# Run a container from the image, mapping port 80 in the container to 8080 on the host
container = client.containers.run(
    image_name,
    detach=True, # Run in the background
    ports={'80/tcp': 8080}
)

print(f"Container '{container.name}' started with ID: {container.short_id}")
# To stop it later, you can run: container.stop()

4. kubernetes (Kubernetes API)

Use Case: Connecting to a Kubernetes cluster (using your local kubeconfig) and listing all the pods in the default namespace.

from kubernetes import client, config

# Load Kubernetes configuration from the default location (~/.kube/config)
config.load_kube_config()

# Create an API client instance
v1 = client.CoreV1Api()

print("Listing pods in the 'default' namespace:")
try:
    pod_list = v1.list_namespaced_pod(namespace="default", watch=False)
    for pod in pod_list.items:
        print(f"Pod: {pod.metadata.name}, Status: {pod.status.phase}, IP: {pod.status.pod_ip}")
except client.ApiException as e:
    print(f"Error listing pods: {e}")

5. pyyaml (YAML Parser)

Use Case: Loading a config.yaml file that contains application settings, such as database credentials, into a Python dictionary.

# Assume you have a file named 'config.yaml' with this content:
#
# app_name: "MyWebApp"
# version: 1.2
# database:
#   host: "db.example.com"
#   user: "admin"
#   port: 5432

import yaml

config_file_path = 'config.yaml'

try:
    with open(config_file_path, 'r') as file:
        config = yaml.safe_load(file)

    # Now you can access the configuration like a dictionary
    db_host = config['database']['host']
    app_version = config['version']

    print(f"Application Name: {config['app_name']}")
    print(f"Connecting to database at: {db_host}")
    print(f"Running version: {app_version}")

except FileNotFoundError:
    print(f"Error: Configuration file '{config_file_path}' not found.")
except yaml.YAMLError as e:
    print(f"Error parsing YAML file: {e}")

6. requests (HTTP Library)

Use Case: Making an HTTP GET request to a service's health check endpoint to verify it's running and responsive.

import requests

# URL of the service's health check endpoint
HEALTH_CHECK_URL = "https://api.example.com/health"

try:
    # Make the request with a 5-second timeout
    response = requests.get(HEALTH_CHECK_URL, timeout=5)

    # Raise an exception for bad status codes (4xx or 5xx)
    response.raise_for_status()

    # If we reach here, the status code was 2xx (e.g., 200 OK)
    print(f"Service is healthy! Status Code: {response.status_code}")
    print("Response JSON:", response.json()) # Assuming the endpoint returns JSON

except requests.exceptions.RequestException as e:
    print(f"Health check failed: {e}")

7. fabric (High-level SSH Task Execution)

Use Case: Defining a simple deployment task in a fabfile.py to pull the latest changes from a git repository on a remote server.

# Create a file named 'fabfile.py'
from fabric import task

@task
def deploy(c):
    """
    Pulls the latest code from the main branch on the remote server.
    To run: fab -H your_server_ip deploy
    """
    print("Connecting to server to deploy...")
    with c.cd("/var/www/my_project"): # Change to the project directory
        print("Pulling latest changes from git...")
        c.run("git pull origin main")
    print("Deployment complete!")

8. pytest (Testing Framework)

Use Case: Writing a simple unit test to ensure a function that formats user data works as expected.

# File: test_formatting.py

# The function we want to test (usually in another file, e.g., utils.py)
def format_user_for_display(first_name, last_name, age):
    if not isinstance(age, int) or age < 0:
        raise ValueError("Age must be a non-negative integer.")
    return f"{last_name.upper()}, {first_name.capitalize()} ({age})"

# The test function for pytest
def test_format_user_for_display():
    # Call the function with sample data
    result = format_user_for_display("anna", "svensson", 34)
    # Assert that the output is what we expect
    assert result == "SVENSSON, Anna (34)"

# To run from your terminal: pytest

9. ansible-runner (Ansible from Python)

Use Case: Programmatically running a simple Ansible playbook from a Python script to ensure nginx is installed on a target host.

# --- Setup ---
# 1. Create a directory 'ansible_project'
# 2. Inside, create an 'inventory' file:
#    [webservers]
#    your_server_ip ansible_user=your_username
# 3. Inside, create a 'playbook.yml' file:
#    ---
#    - hosts: webservers
#      become: yes
#      tasks:
#        - name: Ensure nginx is installed
#          apt:
#            name: nginx
#            state: present

import ansible_runner
import os

private_data_dir = os.path.abspath('./ansible_project')
print(f"Running Ansible playbook from: {private_data_dir}")

# Run the playbook
runner = ansible_runner.run(private_data_dir=private_data_dir, playbook='playbook.yml')

# Check the results
print(f"Playbook finished with status: {runner.status}")
if runner.rc != 0:
    print(f"Error running playbook. Return code: {runner.rc}")
else:
    print("Playbook executed successfully.")

10. sys (System-specific parameters)

Use Case: Creating a simple command-line script that takes a server environment (dev, staging, prod) as an argument to perform an action.

import sys

def main():
    # sys.argv is a list of command-line arguments
    if len(sys.argv) < 2:
        print("Usage: python deploy_script.py <environment>")
        print("Example: python deploy_script.py production")
        sys.exit(1) # Exit with an error code

    environment = sys.argv[1]

    print(f"Starting deployment to the '{environment}' environment...")

    if environment == "production":
        confirm = input("Are you sure you want to deploy to PRODUCTION? (yes/no): ")
        if confirm.lower() != 'yes':
            print("Deployment cancelled.")
            sys.exit(0)

    # ... (add deployment logic here) ...
    print(f"Deployment to '{environment}' completed successfully.")

if __name__ == "__main__":
    main()

# To run from your terminal: python your_script_name.py staging

11. subprocess (Running External Commands)

Use Case: Running a local shell command like terraform plan from within a Python script to automate infrastructure previews.

import subprocess
import os

terraform_dir = "./terraform_project"

if not os.path.isdir(terraform_dir):
    print(f"Directory '{terraform_dir}' not found.")
else:
    print(f"Running 'terraform plan' in {terraform_dir}...")
    # Run the command, capturing its output and decoding it as text
    result = subprocess.run(
        ["terraform", "plan"],
        cwd=terraform_dir,       # Run command in this directory
        capture_output=True,   # Capture stdout and stderr
        text=True              # Decode output as text
    )
    print("--- Terraform Plan Output ---")
    print(result.stdout)
    if result.stderr:
        print("--- Errors ---")
        print(result.stderr)
    print(f"---------------------------\nCommand finished with return code: {result.returncode}")

12. os (Operating System Interface)

Use Case: Securely reading a secret API key from an environment variable instead of hardcoding it in the script.

import os
import requests

# Get API key from an environment variable named 'MY_APP_API_KEY'
# To set it in your terminal (Linux/macOS): export MY_APP_API_KEY='your_secret_key'
api_key = os.getenv("MY_APP_API_KEY")

if not api_key:
    print("Error: MY_APP_API_KEY environment variable not set.")
else:
    print("API Key found. Making an authenticated request...")
    headers = {"Authorization": f"Bearer {api_key}"}
    # Example usage:
    # response = requests.get("https://api.example.com/data", headers=headers)
    # print(f"API Response Status: {response.status_code}")
    print("Request would be made using the secret key.")

13. json (JSON Encoder and Decoder)

Use Case: Parsing a JSON response from an API to extract specific information, such as a software version number.

import json

# Example JSON string from an API response
json_response = '{"service": "inventory-api", "status": "ok", "version": "2.5.1", "components": ["database", "cache"]}'

try:
    # Parse the JSON string into a Python dictionary
    data = json.loads(json_response)

    # Extract specific values safely using .get()
    service_name = data.get("service")
    version = data.get("version")

    print(f"Successfully parsed JSON for service: '{service_name}'")
    print(f"Current running version is: {version}")

except json.JSONDecodeError as e:
    print(f"Failed to decode JSON: {e}")

14. logging (Logging Facility)

Use Case: Setting up structured logging for an automation script to record events to both the console and a file for better debugging and auditing.

import logging

# Configure logging to write to a file and the console
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("automation.log"), # Log to a file
        logging.StreamHandler()                # Log to the console
    ]
)

logging.info("Automation script started.")
try:
    # Simulate a task
    logging.info("Connecting to database...")
    # ... database connection logic ...
    logging.info("Database connection successful.")
    logging.warning("Disk space is running low.")
    # You could uncomment the next line to test error logging
    # raise ValueError("Failed to process item X")
except Exception as e:
    # exc_info=True includes the full traceback in the log
    logging.error(f"An unexpected error occurred: {e}", exc_info=True)
finally:
    logging.info("Automation script finished.")