Exception Bubbling in Python

2023-09-02

A practical approach to exception handling in Python using decorators, demonstrating how to elegantly transform exceptions without modifying method signatures, inspired by Java's explicit exception handling.

One aspect of Java that occasionally nudges at me is its explicit approach to exception handling. Java requires developers to either handle exceptions via try-catch blocks or declare them in method signatures. While it does enforce robustness, it sometimes feels a bit too constrained, especially when compared to the flexible nature of Python.

Recently, I crafted a solution in Python for k8sutils. Instead of the usual explicit exception handling or modifying method signatures, I created a Python decorator - akin to annotations in Java - that substitutes an exception for another without altering the underlying code. Here’s what it looks like:

import functools
import subprocess

def rethrow(exception_type=Exception):
    """Rethrow a CalledProcessError as the specified exception type."""
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except subprocess.CalledProcessError as e:
                raise exception_type(f"Command failed: {e.cmd}. Error: {e.output}") from e
        return wrapper

    return decorator

Using this decorator, it becomes straightforward to alter the exception being thrown:

@rethrow(ValueError)
def get(namespace=None):
    """Get all deployments in the cluster."""
    cmd = "kubectl get deployments -o json"
    if namespace:
        cmd += f" -n {namespace}"

    result = subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
    deployments = json.loads(result.stdout)
    return deployments

The @rethrow(ValueError) decorator automatically translates a CalledProcessError to a ValueError without the need to change the method’s code.

For another example:

@rethrow(RuntimeError)
def delete_deployment(deployment_name):
    """Delete a specific deployment."""
    cmd = f"kubectl delete deployment {deployment_name}"
    subprocess.run(cmd, shell=True, check=True)

Here, instead of bubbling up the generic CalledProcessError, any error encountered will raise a RuntimeError.