When your Python automation script runs unattended at 3 AM, you need to know exactly what happened if something goes wrong. Error handling and logging are not optional — they are the difference between a script that silently fails and one that tells you exactly where, why, and how to fix it. This lesson covers production-grade patterns for exception handling, structured logging, and debugging Python in DevOps environments.

1. Learning Objectives

By the end of this lesson, you will be able to:

  • Implement robust exception handling patterns for Python automation scripts
  • Configure Python's logging module for different environments (dev, staging, production)
  • Write structured JSON logs compatible with log aggregators like Elasticsearch and Loki
  • Use context managers and try/except/finally blocks for resource cleanup
  • Debug failing automation scripts using tracebacks and logging-based diagnostics
  • Build a production-grade monitoring script with proper error reporting

2. Why This Matters

Imagine this: your cron job that backs up databases at 2 AM fails silently for three weeks. When you finally discover the issue, the last three backups are missing. A five-minute investment in proper error handling and logging would have alerted you the very first night.

In DevOps, your scripts run unattended. They deploy code, provision infrastructure, run tests, and monitor systems. When something breaks, you need actionable information immediately. Python's logging module and exception handling give you that visibility. Every major DevOps tool — Ansible, SaltStack, even Kubernetes operators — relies on these same patterns.

3. Core Concepts

Exception Handling Fundamentals

An exception is Python's way of saying "something went wrong." Without handling, exceptions crash your script. With proper handling, you can recover, retry, or log the failure and move on.

try:
    result = 10 / 0
except ZeroDivisionError:
    print('Cannot divide by zero!')
finally:
    print('This always runs — cleanup here.')

# Multiple exception types
try:
    with open('/tmp/config.yaml') as f:
        data = f.read()
except FileNotFoundError:
    print('Config file not found. Using defaults.')
except PermissionError:
    print('Permission denied. Check file ownership.')
except Exception as e:
    print(f'Unexpected error: {e}')
    raise  # Re-raise if you cannot handle it
Basic exception handling structure

The Exception Hierarchy

Python exceptions form a hierarchy rooted in BaseException. The most important branch is Exception — catch this to handle recoverable errors, not SystemExit or KeyboardInterrupt. Common DevOps-relevant exceptions:

  • FileNotFoundError — missing config files, logs, or templates
  • PermissionError — insufficient file or socket permissions
  • ConnectionError — network timeouts to APIs or databases
  • subprocess.CalledProcessError — a shell command returned non-zero
  • json.JSONDecodeError — malformed API responses or config files
  • yaml.YAMLError — invalid YAML in config files or manifests

Python's Logging Module

Python's built-in logging module is the standard for production scripts. It's vastly superior to print() because it provides log levels, timestamps, output routing, and formatting — all configurable without changing your code.

import logging

# Basic configuration (one-liner)
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    filename='/var/log/devops_script.log'  # Remove to log to stderr
)

# Create a logger for your module
logger = logging.getLogger(__name__)

logger.debug('Detailed debug info')   # Only shown when level=DEBUG
logger.info('Script started')          # Normal operational messages
logger.warning('Disk at 85%')          # Something to watch
logger.error('Failed to connect')      # Something failed
logger.critical('Data loss risk')      # Immediate attention
Basic logging setup

Logging Levels Explained

The five standard log levels let you filter noise from signal. In development, set level=DEBUG to see everything. In production, set level=WARNING to see only actionable events. Here is how each level maps to real DevOps scenarios:

  • DEBUG — Detailed variable state, HTTP request/response bodies, timing information. Useful when reproducing a bug.
  • INFO — Routine operations: backup started/finished, deployment initiated, health check passed.
  • WARNING — Something unexpected that does not break the script: disk at 85%, deprecated API version, retry attempt 2 of 3.
  • ERROR — A failure that prevents one operation but the script continues: failed to scrape one server, config parse error using fallback.
  • CRITICAL — The script cannot continue: database connection lost permanently, required config file missing, disk full.

Structured Logging with JSON

Plain-text logs are hard to search and impossible to parse programmatically. Modern DevOps stacks expect structured JSON logs that tools like Elasticsearch, Loki, and CloudWatch can index and query. Here is a custom formatter that outputs JSON:

import json
import logging
from datetime import datetime

class JSONFormatter(logging.Formatter):
    """Format log records as JSON for log aggregators."""
    
    def format(self, record):
        log_entry = {
            'timestamp': datetime.utcfromtimestamp(record.created).isoformat() + 'Z',
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno,
        }
        # Add exception info if present
        if record.exc_info and record.exc_info[0]:
            log_entry['exception'] = {
                'type': record.exc_info[0].__name__,
                'message': str(record.exc_info[1]),
            }
        # Add extra fields passed in the log call
        if hasattr(record, 'extra_fields'):
            log_entry.update(record.extra_fields)
        return json.dumps(log_entry, default=str)

# Usage
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger = logging.getLogger('deployment')
logger.addHandler(handler)
logger.setLevel(logging.INFO)

logger.info('Deployment started', extra={
    'extra_fields': {'service': 'api', 'version': '2.5.0', 'env': 'production'}
})
JSON formatter for cloud-native logging

Context Managers for Resource Cleanup

Context managers (the with statement) guarantee cleanup even when exceptions occur. Always use them for file handles, network connections, database sessions, and subprocesses.

# Without context manager — easy to leak resources
try:
    f = open('/var/log/app.log')
    data = f.read()
    # If something raises here, f never closes
except Exception:
    f.close()  # This line never runs if 'data = f.read()' fails
    raise

# With context manager — always safe
with open('/var/log/app.log') as f:
    data = f.read()  # File closes automatically, even on exception

# Custom context manager for SSH connections
from contextlib import contextmanager

@contextmanager
def ssh_connection(hostname, user):
    """Context manager for SSH connections with auto-cleanup."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(hostname, username=user)
        logger.info(f'Connected to {hostname}')
        yield client  # Code inside 'with' runs here
    finally:
        client.close()
        logger.info(f'Disconnected from {hostname}')

# Usage
with ssh_connection('web-01', 'deploy') as ssh:
    stdin, stdout, stderr = ssh.exec_command('docker ps')
Context managers for safe resource handling

4. Hands-On Practice: Building a Health Check Monitor

Let us build a production-ready health check monitor that pings multiple services, logs structured results, and handles every failure mode. This is the kind of tool you would run as a cron job or Kubernetes sidecar:

#!/usr/bin/env python3
"""
health_monitor.py - Production health check with structured logging.
Checks HTTP endpoints, captures metrics, and reports failures.
"""
import json
import logging
import sys
import time
import urllib.request
import urllib.error
from datetime import datetime

# --- JSON Logging Setup ---

class JSONFormatter(logging.Formatter):
    def format(self, record):
        entry = {
            'timestamp': datetime.utcfromtimestamp(record.created).isoformat() + 'Z',
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
        }
        if record.exc_info and record.exc_info[0]:
            entry['exception'] = record.exc_info[1].__class__.__name__
        if hasattr(record, 'extra_fields'):
            entry.update(record.extra_fields)
        return json.dumps(entry)

handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logger = logging.getLogger('health-monitor')
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# --- Configuration ---

SERVICES = [
    {'name': 'API Gateway', 'url': 'https://api.example.com/health'},
    {'name': 'Database',    'url': 'https://db.example.com/healthz'},
    {'name': 'Cache',       'url': 'https://redis.example.com/ping'},
    {'name': 'Auth',        'url': 'https://auth.example.com/ready'},
]

TIMEOUT = 10  # seconds
RETRIES = 2
RETRY_DELAY = 2  # seconds

# --- Core Logic ---

def check_endpoint(name, url, timeout=TIMEOUT):
    """Check a single HTTP endpoint with retry logic."""
    for attempt in range(1, RETRIES + 1):
        try:
            start = time.time()
            req = urllib.request.Request(url)
            with urllib.request.urlopen(req, timeout=timeout) as resp:
                duration = time.time() - start
                status = resp.status
                body = resp.read().decode('utf-8')[:200]
            
            is_healthy = 200 <= status < 400
            result = {
                'service': name, 'url': url,
                'status': 'UP' if is_healthy else 'DEGRADED',
                'http_status': status, 'duration_ms': round(duration * 1000),
            }
            
            if is_healthy:
                logger.info(f'{name} is healthy', extra={'extra_fields': result})
            else:
                logger.warning(f'{name} returned {status}', extra={'extra_fields': result})
            
            return result
            
        except urllib.error.HTTPError as e:
            logger.warning(f'{name} HTTP {e.code} (attempt {attempt}/{RETRIES})',
                          extra={'extra_fields': {'service': name, 'attempt': attempt}})
            if attempt < RETRIES:
                time.sleep(RETRY_DELAY)
                
        except (urllib.error.URLError, ConnectionError, TimeoutError) as e:
            logger.error(f'{name} connection failed: {e} (attempt {attempt}/{RETRIES})',
                        extra={'extra_fields': {'service': name, 'attempt': attempt}})
            if attempt < RETRIES:
                time.sleep(RETRY_DELAY)
    
    # All retries exhausted
    result = {
        'service': name, 'url': url,
        'status': 'DOWN', 'http_status': None, 'duration_ms': None,
    }
    logger.critical(f'{name} is DOWN after {RETRIES} retries',
                   extra={'extra_fields': result})
    return result

def run_health_checks():
    """Run health checks on all configured services."""
    logger.info('Health check cycle starting',
               extra={'extra_fields': {'service_count': len(SERVICES)}})
    
    results = []
    failures = 0
    
    for svc in SERVICES:
        result = check_endpoint(svc['name'], svc['url'])
        results.append(result)
        if result['status'] == 'DOWN':
            failures += 1
    
    summary = {
        'total': len(SERVICES),
        'up': len(SERVICES) - failures,
        'down': failures,
        'healthy_pct': round((len(SERVICES) - failures) / len(SERVICES) * 100),
    }
    
    if failures:
        logger.warning(f'Health summary: {summary["up"]}/{summary["total"]} UP',
                      extra={'extra_fields': {'summary': summary}})
    else:
        logger.info(f'All services healthy ({summary["up"]}/{summary["total"]})',
                   extra={'extra_fields': {'summary': summary}})
    
    return summary

if __name__ == '__main__':
    try:
        summary = run_health_checks()
        sys.exit(0 if summary['down'] == 0 else 1)
    except Exception as e:
        logger.critical('Health monitor crashed',
                       extra={'extra_fields': {'error': str(e)}})
        sys.exit(2)
Complete production health monitor script

This script demonstrates every concept from this lesson: structured JSON logging, retry logic with exponential fallback, exception-specific handling, context managers for HTTP connections, and a non-zero exit code that cron or Kubernetes can detect.

5. Common Errors & Solutions

  • Error: "No handlers could be found for logger" — You created a logger but never called basicConfig() or added a handler. Always configure logging before creating loggers.
  • Error: Logs go to stderr but you want stdout — By default basicConfig() sends to stderr. Add stream=sys.stdout to send logs to stdout instead of stderr.
  • Error: "Exception ignored" in context manager — Your __exit__ method raised an exception while handling another exception. Keep cleanup code simple — use finally or the contextlib decorator.
  • Error: JSON logs are empty objects — The json.dumps() call failed because your log entry contains a non-serializable type like datetime. Add default=str to json.dumps() as shown in the formatter above.
  • Error: Multiple log entries per message — You called basicConfig() multiple times, adding duplicate handlers. Check for existing handlers before adding: if not logger.handlers: logger.addHandler(handler).

6. Summary Checklist

  • Understand Python exception hierarchy and catch specific exceptions, not bare except:
  • Use logging.basicConfig() to set up module-level logging in one line
  • Choose appropriate log levels: DEBUG for development, INFO for operations, WARNING for anomalies, ERROR for failures, CRITICAL for unrecoverable
  • Implement structured JSON logging using a custom formatter for log aggregator compatibility
  • Always use context managers (with) for files, network connections, and subprocesses
  • Implement retry logic with backoff for transient failures (network timeouts, 503 errors)
  • Use logging.getLogger(__name__) per module for hierarchical logger names
  • Set log level via environment variable: LOG_LEVEL=DEBUG python3 script.py

7. Practice Exercise

Build a log tailer with alerting: Write a Python script that tails a log file (like /var/log/syslog), watches for ERROR and CRITICAL entries, and outputs structured JSON alerts. Requirements:

  • Use try/except to handle the file not existing yet (wait and retry)
  • Use a context manager to open the file
  • Parse each line and output JSON with: timestamp, raw message, severity level, and source process
  • If 5+ errors appear within 60 seconds, log a CRITICAL alert about error burst
  • Handle KeyboardInterrupt gracefully — log a shutdown message and exit cleanly

Bonus: Add a --json-output flag using argparse or click (from the previous lesson) that switches between human-readable and JSON output.

8. Next Steps

You now have the skills to make your Python scripts production-ready with proper error handling and structured logging. Your automation will no longer fail silently — every error will be captured, formatted, and routed to your logging infrastructure.

In the next lesson, we will build on this foundation by exploring Python for DevOps: Working with Databases and Persistent Storage. You will learn to connect to PostgreSQL and Redis from Python, manage connection pools, handle transaction failures gracefully, and build data migration scripts that log every step.

Tags: python, error-handling, logging, debugging, automation, devops, monitoring

Gataya Med

DevOps Engineer & Backend Developer. Sharing insights on cloud, automation, and scalable systems.

gatayaMed View Profile @https://github.com/gatayaMed

Comments (0)

Sarah Chen July 23, 2026

This is exactly what I needed! The initContainer approach solved our migration issues completely. Thanks for the detailed guide!

Reply

Leave a Comment