Python for DevOps and Automation

April 2, 2026 21 min read DevOps

pythondevopsautomationscriptingapisansibleboto3

Contents

Python is the lingua franca of DevOps. Not because it’s the fastest language or the most elegant, but because it sits at the intersection of readable, powerful, and universally available. Ansible is written in Python. The AWS, GCP, and Azure SDKs all have first-class Python support. Most observability tooling either runs on Python or provides Python clients first. Kubernetes operators are increasingly written in Python using frameworks like Kopf. If you’re working in ops, you will write Python.

This post covers the Python concepts, libraries, and patterns that actually matter for automation work—not beginner tutorials about variables and loops, but the things you need to write production-quality ops scripts.

Why Python Dominates Ops Work

A few concrete reasons:

Speed of iteration matters more than execution speed. Most ops scripts are I/O-bound—waiting for API calls, SSH connections, file operations. CPython’s overhead is irrelevant when you’re waiting on network round-trips.

Ecosystem depth is unmatched. Nearly every infrastructure API has a Python SDK. If there’s no SDK, someone has written a Python client library. If there’s no client library, requests gets you there in 10 lines.

Scripts read like documentation. Python’s syntax is close enough to pseudocode that a script explaining a deployment process is also documenting it.

Cross-platform. The same script runs on Linux, macOS, and Windows without modification. Critical for teams using different workstations.

The practical result: you’ll encounter more Python in ops tooling than any other language, so reading and writing it fluently pays dividends across the entire job.

Core Concepts Ops Engineers Actually Use

These aren’t beginner concepts—they’re the specific Python features that come up constantly in ops code.

List Comprehensions and Generator Expressions

The baseline for processing API responses, log files, and configuration data:

1
2
3
4
5
6
7
8
9


# Filter unhealthy hosts from an API response
unhealthy = [h['name'] for h in hosts if h['status'] != 'healthy']

# Build a dict of name → IP for prod hosts
host_ips = {h['name']: h['ip'] for h in hosts if 'prod' in h['tags']}

# Generator expression—lazy evaluation for large log files
for line in (l.strip() for l in open('access.log') if 'ERROR' in l):
    process_error(line)

The distinction between list comprehensions [...] and generator expressions (...) matters for large datasets. List comprehensions build the entire list in memory. Generator expressions yield one item at a time. When piping to another function or iterating once, prefer generators—they use constant memory regardless of input size.

Context Managers

The with statement ensures cleanup happens even if exceptions are thrown. This matters enormously in ops code dealing with file handles, database connections, SSH sessions, and temporary directories:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


# SSH connection closes even if the command raises
with paramiko.SSHClient() as ssh:
    ssh.connect("10.0.1.10", username="admin")
    stdin, stdout, stderr = ssh.exec_command("systemctl status nginx")
    for line in stdout:
        print(line.strip())
# ssh.close() called automatically

# Multiple resources—both cleaned up on exit
with open('config.yaml') as f, tempfile.TemporaryDirectory() as tmpdir:
    config = yaml.safe_load(f)
    # tmpdir deleted on exit even if processing fails

You can write your own context managers for common ops patterns like timing operations:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


from contextlib import contextmanager
import time

@contextmanager
def timer(label):
    start = time.time()
    try:
        yield
    finally:
        elapsed = time.time() - start
        print(f"[{label}] {elapsed:.2f}s")

with timer("API sync"):
    results = sync_all_deployments()

Generators and `yield`

Generators shine when processing streams of data—paginated API results, log files, remote command output:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


def fetch_all_pages(session, url, params=None):
    """Yield items from a paginated API without loading all pages at once."""
    page = 1
    while True:
        resp = session.get(url, params={**(params or {}), 'page': page, 'limit': 100})
        resp.raise_for_status()
        data = resp.json()

        items = data.get('items') or data.get('data') or data.get('results', [])
        if not items:
            break

        yield from items

        if not data.get('hasMore'):
            break
        page += 1

# Process 10,000 servers without holding them all in memory
for server in fetch_all_pages(session, "https://api.example.com/servers"):
    if server['status'] == 'degraded':
        alert(server)

Decorators

Decorators let you add behavior (retries, logging, timing) to functions without cluttering their implementation:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


import time
from functools import wraps

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

@retry(max_attempts=5, delay=1, backoff=2, exceptions=(ConnectionError, TimeoutError))
def fetch_deployment_status(api_client, deployment_id):
    return api_client.get(f'/deployments/{deployment_id}')

For production use, prefer the tenacity library over rolling your own retry logic—it handles more edge cases.

Dataclasses

@dataclass replaces boilerplate-heavy __init__ methods and provides free __repr__, __eq__, and dict serialization:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25


from dataclasses import dataclass, field, asdict
from typing import List, Optional

@dataclass
class Host:
    name: str
    ip: str
    port: int = 22
    username: str = "ansible"
    tags: List[str] = field(default_factory=list)

    def ssh_url(self) -> str:
        return f"{self.username}@{self.ip}:{self.port}"

@dataclass
class DeploymentConfig:
    service_name: str
    version: str
    replicas: int = 3
    cpu_limit: Optional[str] = None
    memory_limit: Optional[str] = None

web01 = Host(name="web01", ip="10.0.1.10", tags=["prod", "frontend"])
print(web01.ssh_url())    # ansible@10.0.1.10:22
print(asdict(web01))      # dict for JSON serialization

For configuration with validation (required fields, type coercion, range checks), upgrade to pydantic v2 instead of plain dataclasses.

pathlib: Never Use `os.path` Again

pathlib.Path makes file operations readable and cross-platform:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24


from pathlib import Path
import yaml

# Build paths safely
config_file = Path.home() / ".config" / "myapp" / "config.yaml"

if config_file.exists():
    config = yaml.safe_load(config_file.read_text())

# Find all log files recursively
for log in Path("/var/log").rglob("*.log"):
    if log.stat().st_size > 100_000_000:  # > 100MB
        print(f"Large log: {log} ({log.stat().st_size / 1e6:.0f}MB)")

# Create directories and write atomically
output = Path("/var/reports/daily")
output.mkdir(parents=True, exist_ok=True)
(output / "report.json").write_text(json.dumps(data, indent=2))

# Path operations
log_file = Path("/var/log/app.log")
print(log_file.parent)   # /var/log
print(log_file.stem)     # app
print(log_file.suffix)   # .log

Working with APIs

requests: The Standard

Use requests for synchronous HTTP. Always configure timeouts and retry adapters:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34


import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session(token: str) -> requests.Session:
    """Create a resilient session with built-in retries."""
    session = requests.Session()
    session.headers.update({"Authorization": f"Bearer {token}"})

    retry = Retry(
        total=3,
        backoff_factor=1,                              # 1s, 2s, 4s delays
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST", "PUT", "DELETE"]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session

session = create_session(API_TOKEN)

try:
    resp = session.get(
        "https://api.example.com/deployments",
        params={"status": "active", "env": "prod"},
        timeout=10        # ALWAYS set this
    )
    resp.raise_for_status()    # Raise on 4xx/5xx
    deployments = resp.json()
except requests.exceptions.Timeout:
    logger.error("API timed out")
except requests.exceptions.HTTPError as e:
    logger.error(f"HTTP {e.response.status_code}: {e.response.text}")

For streaming large responses (log archives, file downloads):

1
2
3
4


resp = session.get("https://api.example.com/logs/export", stream=True)
with open("export.tar.gz", "wb") as f:
    for chunk in resp.iter_content(chunk_size=8192):
        f.write(chunk)

httpx: Async HTTP

When you need to make many concurrent API calls, httpx provides the same interface as requests but with async/await support:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


import httpx
import asyncio

async def check_all_endpoints(urls: list[str]) -> dict[str, bool]:
    """Check health of many endpoints concurrently."""
    async with httpx.AsyncClient(timeout=5.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

    results = {}
    for url, resp in zip(urls, responses):
        if isinstance(resp, Exception):
            results[url] = False
        else:
            results[url] = resp.status_code == 200
    return results

# Sequential: 50 endpoints × 1s each = 50 seconds
# Concurrent: 50 endpoints × 1s each ≈ 1-2 seconds
health = asyncio.run(check_all_endpoints(endpoint_urls))

Retries with tenacity

For more sophisticated retry logic than Retry from urllib3:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35


from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    retry_if_result,
    before_sleep_log,
)
import logging

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((ConnectionError, TimeoutError)),
    before_sleep=before_sleep_log(logger, logging.WARNING),
    reraise=True
)
def deploy_to_cluster(service, version):
    """Deploy with exponential backoff on transient failures."""
    # ...

# Retry until a condition is met (polling pattern)
@retry(
    stop=stop_after_attempt(60),          # max 60 attempts
    wait=wait_exponential(min=5, max=30), # poll every 5-30s
    retry=retry_if_result(lambda status: status != 'ready')
)
def wait_for_deployment(deployment_id):
    """Wait for deployment to become ready."""
    resp = requests.get(f"https://api.example.com/deployments/{deployment_id}")
    return resp.json()['status']

status = wait_for_deployment("deploy-abc123")

File and System Operations

subprocess: Running Commands

subprocess.run() is the right choice for most command execution:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30


import subprocess
import json

# Capture output
result = subprocess.run(
    ["docker", "ps", "--format", "{{json .}}"],
    capture_output=True,
    text=True,
    timeout=10
)
if result.returncode != 0:
    logger.error(f"docker ps failed: {result.stderr}")
    raise RuntimeError("Docker unavailable")

# Each line is a JSON object (docker's streaming format)
containers = [json.loads(line) for line in result.stdout.strip().split('\n') if line]

# Stream output for long-running commands
with subprocess.Popen(
    ["ansible-playbook", "site.yml", "-i", "inventory.ini"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    bufsize=1
) as proc:
    for line in proc.stdout:
        print(line, end='')           # Real-time output
    proc.wait()
    if proc.returncode != 0:
        raise RuntimeError(f"Playbook failed with exit code {proc.returncode}")

Use subprocess.Popen instead of run() when you need bidirectional communication (write to stdin while reading stdout) or real-time streaming. Avoid shell=True unless you’re piping multiple commands—it opens injection vulnerabilities if any input comes from user data.

paramiko and Fabric for SSH

paramiko is the low-level SSH library; Fabric wraps it in a cleaner interface:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


from fabric import Connection

# Single host
with Connection("10.0.1.10", user="admin") as conn:
    result = conn.run("systemctl status nginx", hide=True)
    if "active (running)" not in result.stdout:
        conn.run("systemctl start nginx")

    # Upload file
    conn.put("/tmp/new-config.yaml", "/etc/myapp/config.yaml")
    conn.run("systemctl reload myapp")

# Run across multiple hosts
from fabric import ThreadingGroup

pool = ThreadingGroup(
    "web01", "web02", "web03",
    user="admin",
    connect_kwargs={"key_filename": "~/.ssh/deploy_key"}
)
results = pool.run("uptime")
for conn, result in results.items():
    print(f"{conn.host}: {result.stdout.strip()}")

Configuration Management

YAML

YAML is the config format of the cloud-native world. Use yaml.safe_load()—never yaml.load(), which can execute arbitrary code:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


import yaml
from pathlib import Path

# Reading
config = yaml.safe_load(Path("config.yaml").read_text())

# Writing
data = {
    "all": {
        "hosts": {
            "web01": {"ansible_host": "10.0.1.10"},
            "web02": {"ansible_host": "10.0.1.11"},
        }
    }
}
with open("inventory.yaml", "w") as f:
    yaml.dump(data, f, default_flow_style=False)

When you need to modify a YAML file without destroying comments or reformatting, use ruamel.yaml:

1
2
3
4
5
6
7
8


from ruamel.yaml import YAML

yaml = YAML()
yaml.preserve_quotes = True

config = yaml.load(Path("values.yaml"))
config['image']['tag'] = '1.5.2'          # Update a field
yaml.dump(config, Path("values.yaml"))    # Write back preserving structure

TOML

Python 3.11+ includes tomllib for reading TOML (e.g., pyproject.toml). For writing, install tomli_w:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


import tomllib
import tomli_w
from pathlib import Path

# Read
config = tomllib.loads(Path("pyproject.toml").read_text())
version = config['project']['version']

# Write
Path("release.toml").write_bytes(tomli_w.dumps({
    "version": version,
    "built_at": "2026-04-02T00:00:00Z"
}))

Environment Variables and Secrets

Never hardcode credentials in scripts. Load them from environment variables:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


import os
from dotenv import load_dotenv

load_dotenv("/etc/myapp/.env")            # Load from .env file

API_TOKEN = os.getenv("API_TOKEN")
if not API_TOKEN:
    raise RuntimeError("API_TOKEN environment variable not set")

DB_URL = os.getenv("DATABASE_URL", "postgresql://localhost/myapp")

For applications with many config values, use pydantic-settings:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11


from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    api_token: str               # Required—fails startup if missing
    api_url: str = "https://api.example.com"
    debug: bool = False
    replicas: int = 3

    model_config = {"env_file": "/etc/myapp/.env"}

settings = Settings()            # Validates types, raises on missing required fields

Building CLI Tools

Three options, in order of increasing complexity:

argparse (no dependencies, good for simple scripts):

1
2
3
4
5
6
7
8


import argparse

parser = argparse.ArgumentParser(description="Deploy a service")
parser.add_argument("service", help="Service name")
parser.add_argument("--version", required=True)
parser.add_argument("--replicas", type=int, default=3)
parser.add_argument("--dry-run", action="store_true")
args = parser.parse_args()

Click (most widely used, excellent for complex CLIs with subcommands):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25


import click

@click.group()
def cli():
    pass

@cli.command()
@click.argument("service")
@click.option("--version", required=True)
@click.option("--replicas", default=3)
@click.option("--dry-run", is_flag=True)
def deploy(service, version, replicas, dry_run):
    """Deploy a service to the cluster."""
    click.echo(f"Deploying {service} v{version} ({replicas} replicas)")
    if dry_run:
        click.secho("DRY RUN — no changes applied", fg='yellow')

@cli.command()
@click.argument("service")
def rollback(service):
    """Roll back a service to the previous version."""
    click.echo(f"Rolling back {service}...")

if __name__ == "__main__":
    cli()

Typer (modern, uses type hints, least boilerplate):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18


import typer

app = typer.Typer()

@app.command()
def deploy(
    service: str,
    version: str = typer.Option(..., help="Version to deploy"),
    replicas: int = typer.Option(3),
    dry_run: bool = typer.Option(False, "--dry-run"),
):
    """Deploy a service to the cluster."""
    typer.echo(f"Deploying {service} v{version}...")
    if dry_run:
        typer.secho("DRY RUN", fg=typer.colors.YELLOW)

if __name__ == "__main__":
    app()

Recommendation: Use Typer for new scripts (cleanest syntax). Use Click if you need complex subcommands or are joining an existing codebase that uses it. Use argparse only when you have a strict “no third-party dependencies” constraint.

Logging Setup

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


import logging
import logging.handlers
from pathlib import Path

def setup_logging(name: str, log_file: str = None, level=logging.INFO):
    logger = logging.getLogger(name)
    logger.setLevel(logging.DEBUG)   # Logger accepts everything; handlers filter

    fmt = logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")

    # Console: INFO and above
    console = logging.StreamHandler()
    console.setLevel(level)
    console.setFormatter(fmt)
    logger.addHandler(console)

    # File: DEBUG and above, with rotation
    if log_file:
        Path(log_file).parent.mkdir(parents=True, exist_ok=True)
        file_handler = logging.handlers.RotatingFileHandler(
            log_file, maxBytes=10_000_000, backupCount=5
        )
        file_handler.setLevel(logging.DEBUG)
        file_handler.setFormatter(fmt)
        logger.addHandler(file_handler)

    return logger

logger = setup_logging("deploy", log_file="/var/log/deploy.log")

Cloud SDK Fundamentals

Boto3 (AWS)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33


import boto3
from botocore.exceptions import ClientError

# EC2: find all running prod instances
ec2 = boto3.client('ec2', region_name='us-east-1')
response = ec2.describe_instances(
    Filters=[
        {'Name': 'tag:Environment', 'Values': ['prod']},
        {'Name': 'instance-state-name', 'Values': ['running']},
    ]
)
for reservation in response['Reservations']:
    for instance in reservation['Instances']:
        print(f"{instance['InstanceId']} — {instance['PrivateIpAddress']}")

# S3: upload a file
s3 = boto3.client('s3')
s3.upload_file('/tmp/backup.tar.gz', 'my-backups', 'daily/backup.tar.gz')

# SSM Parameter Store: retrieve secrets without hardcoding
ssm = boto3.client('ssm')
param = ssm.get_parameter(Name='/prod/db/password', WithDecryption=True)
db_password = param['Parameter']['Value']

# Handle AWS API errors
try:
    ec2.start_instances(InstanceIds=['i-1234567890'])
except ClientError as e:
    error_code = e.response['Error']['Code']
    if error_code == 'IncorrectInstanceState':
        print("Instance already running")
    else:
        raise

Boto3 has two interfaces: the client interface (low-level, maps 1:1 to API calls) and the resource interface (higher-level, object-oriented). The client interface is more explicit and generally preferred for scripts.

Google Cloud

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


from google.cloud import storage
from google.auth import default

credentials, project = default()

# GCS operations
storage_client = storage.Client(project=project)
bucket = storage_client.bucket('my-bucket')

# Upload
blob = bucket.blob('backups/daily.tar.gz')
blob.upload_from_filename('/tmp/backup.tar.gz')

# List objects with a prefix
for blob in bucket.list_blobs(prefix='backups/'):
    print(f"{blob.name}: {blob.size} bytes")

Azure

1
2
3
4
5
6
7
8
9


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

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

# List all VMs in a resource group
for vm in compute.virtual_machines.list(resource_group_name):
    print(f"{vm.name} — {vm.location}")

DefaultAzureCredential automatically chains through multiple authentication methods (managed identity, environment variables, CLI login). Use it instead of hardcoded credentials.

Testing Automation Scripts

Automation scripts are often undertested because they’re “just scripts.” This causes pain. A broken deployment script at 2am is not fun. Test the logic.

pytest Basics

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29


# test_deploy.py
import pytest
from unittest.mock import patch, Mock
from pathlib import Path

# Test with mocked HTTP calls
@patch('myapp.deploy.requests.post')
def test_deploy_success(mock_post):
    mock_post.return_value.json.return_value = {'id': 'deploy-123', 'status': 'success'}
    mock_post.return_value.raise_for_status = Mock()

    from myapp.deploy import create_deployment
    result = create_deployment('api-service', '1.0.0')

    assert result['id'] == 'deploy-123'
    mock_post.assert_called_once()

# Fixture for temp files
@pytest.fixture
def config_file(tmp_path):
    f = tmp_path / "config.yaml"
    f.write_text("service: test\nreplicas: 3\n")
    return f

def test_load_config(config_file):
    from myapp.config import load_config
    config = load_config(config_file)
    assert config['service'] == 'test'
    assert config['replicas'] == 3

Mocking subprocess and SSH

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19


@patch('subprocess.run')
def test_docker_check(mock_run):
    mock_run.return_value = Mock(returncode=0, stdout='{"Names":"nginx","Status":"Up 2 hours"}')
    from myapp.docker import get_containers
    containers = get_containers()
    assert len(containers) == 1

@patch('paramiko.SSHClient')
def test_remote_command(mock_ssh_class):
    mock_ssh = mock_ssh_class.return_value.__enter__.return_value
    mock_ssh.exec_command.return_value = (
        None,
        iter(["active (running)\n"]),
        iter([])
    )

    from myapp.ssh import check_service_status
    status = check_service_status("10.0.1.1", "nginx")
    assert status == "active"

Performance: When Python is Fast Enough (and When It Isn’t)

Most ops automation is I/O-bound. You’re waiting on:

API response times (50–500ms per call)
SSH round-trips (10–100ms)
File system operations

In these scenarios Python’s runtime overhead is irrelevant. A 100ms API call takes 100ms whether it’s Python or Go.

Python struggles with:

CPU-bound processing of large datasets (use NumPy, Polars, or rewrite in Go/Rust)
High-concurrency servers handling thousands of simultaneous connections
Startup time in Lambda/serverless where cold starts matter

For the common case of many concurrent I/O operations, asyncio provides dramatic speedups:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


import asyncio
import httpx

async def check_all_services(service_urls: list[str]) -> dict[str, bool]:
    """Check 100 services in ~1 second instead of ~100 seconds."""
    async with httpx.AsyncClient(timeout=5.0) as client:
        tasks = [client.get(url) for url in service_urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

    return {
        url: not isinstance(resp, Exception) and resp.status_code == 200
        for url, resp in zip(service_urls, responses)
    }

results = asyncio.run(check_all_services(all_service_endpoints))

For CPU-bound parallelism:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor

# CPU-bound: analyze log chunks in parallel processes
def analyze_large_file(path: str) -> dict:
    with ProcessPoolExecutor(max_workers=4) as executor:
        chunks = read_in_chunks(path, chunk_size=10_000_000)
        results = list(executor.map(analyze_chunk, chunks))
    return aggregate(results)

# I/O-bound but not async: run SSH commands on multiple hosts in parallel threads
def patch_all_hosts(hosts: list[str]) -> dict:
    with ThreadPoolExecutor(max_workers=20) as executor:
        futures = {executor.submit(patch_host, h): h for h in hosts}
        return {host: future.result() for future, host in futures.items()}

Real-World Example: Deployment Health Report

Here’s a complete script that polls a REST API, aggregates deployment health, and writes a structured report. It demonstrates the patterns above working together:

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143


#!/usr/bin/env python3
"""
health_report.py — Generate a deployment health report.

Usage:
    health_report.py --api-url https://api.example.com --output /tmp/report.json
"""

import asyncio
import json
import logging
import logging.handlers
import sys
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
from typing import List

import httpx
import typer
from tenacity import retry, stop_after_attempt, wait_exponential

# --- Logging Setup ---

def setup_logging(level=logging.INFO):
    logger = logging.getLogger("health_report")
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler()
    handler.setLevel(level)
    handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
    logger.addHandler(handler)
    return logger

logger = setup_logging()

# --- Data Model ---

@dataclass
class DeploymentStatus:
    name: str
    version: str
    desired_replicas: int
    ready_replicas: int
    status: str

    @property
    def health(self) -> str:
        if self.ready_replicas == self.desired_replicas:
            return "healthy"
        elif self.ready_replicas > 0:
            return "degraded"
        else:
            return "unhealthy"

# --- API Client ---

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    reraise=True,
)
async def fetch_deployments(
    api_url: str, api_token: str
) -> List[DeploymentStatus]:
    deployments = []
    async with httpx.AsyncClient(
        headers={"Authorization": f"Bearer {api_token}"},
        timeout=30.0,
    ) as client:
        page = 1
        while True:
            resp = await client.get(
                f"{api_url}/v1/deployments",
                params={"page": page, "limit": 100},
            )
            resp.raise_for_status()
            data = resp.json()

            for item in data.get("items", []):
                deployments.append(DeploymentStatus(
                    name=item["name"],
                    version=item["version"],
                    desired_replicas=item["spec"]["replicas"],
                    ready_replicas=item["status"].get("readyReplicas", 0),
                    status=item["status"]["phase"],
                ))

            if not data.get("hasMore"):
                break
            page += 1

    return deployments

# --- Report Generation ---

def write_report(deployments: List[DeploymentStatus], output: Path):
    unhealthy = [d for d in deployments if d.health != "healthy"]
    report = {
        "generated_at": datetime.utcnow().isoformat() + "Z",
        "total": len(deployments),
        "healthy": len(deployments) - len(unhealthy),
        "unhealthy_count": len(unhealthy),
        "unhealthy": [asdict(d) for d in unhealthy],
        "all": [asdict(d) for d in deployments],
    }
    output.write_text(json.dumps(report, indent=2))
    logger.info(f"Report: {len(deployments)} total, {len(unhealthy)} unhealthy → {output}")
    return unhealthy

# --- CLI ---

app = typer.Typer()

@app.command()
def main(
    api_url: str = typer.Option(..., envvar="API_URL"),
    api_token: str = typer.Option(..., envvar="API_TOKEN"),
    output: Path = typer.Option(Path("/tmp/health_report.json")),
    fail_on_unhealthy: bool = typer.Option(False, "--fail-on-unhealthy"),
):
    """Fetch deployment statuses and generate a health report."""
    try:
        deployments = asyncio.run(fetch_deployments(api_url, api_token))
        unhealthy = write_report(deployments, output)

        if unhealthy:
            typer.secho(f"\n{len(unhealthy)} unhealthy deployments:", fg=typer.colors.RED)
            for d in unhealthy:
                typer.echo(f"  - {d.name} v{d.version}: {d.health} "
                           f"({d.ready_replicas}/{d.desired_replicas} ready)")

        if fail_on_unhealthy and unhealthy:
            raise typer.Exit(code=1)

    except httpx.HTTPStatusError as e:
        logger.error(f"API error {e.response.status_code}: {e.response.text}")
        raise typer.Exit(code=2)
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise typer.Exit(code=1)

if __name__ == "__main__":
    app()

Run it in CI to gate deployments: python health_report.py --fail-on-unhealthy returns exit code 1 if anything is unhealthy, failing the pipeline.

Packaging and Distribution

For scripts you share across a team or run in CI, package them properly:

pyproject.toml (the modern standard):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21


[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "ops-tools"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "httpx>=0.28",
    "typer>=0.13",
    "pydantic-settings>=2.0",
    "tenacity>=8.3",
    "boto3>=1.35",
    "pyyaml>=6.0",
    "python-dotenv>=1.0",
]

[project.scripts]
health-report = "ops_tools.health_report:app"
deploy = "ops_tools.deploy:app"

uv for package management (faster than pip, handles virtualenvs, lock files):

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13


# Install uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Create project
uv init ops-tools
uv add httpx typer tenacity

# Run a script
uv run health_report.py

# Lock and sync for reproducible CI
uv lock
uv sync

pipx for installing CLI tools in isolated environments:

1
2


pipx install ops-tools            # Installs to isolated venv, adds to PATH
pipx upgrade ops-tools

Modern Python Ops Stack (2026)

Concern	Tool	Notes
Runtime	Python 3.12 or 3.13	3.13 has faster startup
Package manager	uv	Replaces pip + virtualenv + pip-tools
HTTP (sync)	requests + tenacity	Standard for scripts
HTTP (async)	httpx	For concurrent API calls
CLI	Typer	Cleanest; falls back to Click
Config validation	pydantic v2	Type-safe settings
AWS	boto3	No replacement
SSH/Remote	Fabric 3.x	Wraps paramiko
Task runner	Invoke	Makefile alternative
Testing	pytest 8.x	Add pytest-asyncio for async tests
Type checking	mypy	Catches bugs before runtime
Linting	ruff	Replaces flake8 + isort + pyupgrade
Structured logging	structlog	Better than stdlib for JSON logs

The shift from pip to uv is the biggest workflow change in recent years. It’s 10-100x faster than pip for installs, manages virtualenvs automatically, and produces a lock file for reproducible environments. If you’re still using bare pip in 2026, it’s worth switching.

Quick Reference: Things That Will Save You

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27


# Don't catch bare Exception—always be specific
try:
    do_something()
except (ConnectionError, TimeoutError) as e:    # Good
    handle_network_error(e)

# Avoid shell=True with user input—injection risk
# Bad: subprocess.run(f"kubectl apply -f {user_input}", shell=True)
# Good:
subprocess.run(["kubectl", "apply", "-f", filename])

# Use walrus operator (:=) for assignment in conditions (Python 3.8+)
while chunk := f.read(8192):
    process(chunk)

# f-strings support expressions
print(f"Disk usage: {shutil.disk_usage('/').used / 1e9:.1f}GB")

# Type hints make scripts self-documenting
def deploy(service: str, version: str, replicas: int = 3) -> bool:
    ...

# sys.exit() with integer codes is important for CI
import sys
sys.exit(0)   # success
sys.exit(1)   # generic failure
sys.exit(2)   # usage/config error

Python’s value in ops work isn’t about language elegance—it’s about leverage. One well-written script with proper error handling, retry logic, and a clean CLI interface can eliminate hours of manual work and serve a team indefinitely. Write it carefully the first time.