Photo by author
# Introduction
Error handling is often a weak point in otherwise solid code. Problems like missing keys, failed requests, and long-running functions appear frequently in real projects. Python built-in try-except Blocks are useful, but they don’t cover many practical cases by themselves.
You’ll need to wrap common failure scenarios in small, reusable functions that help handle retries with limits, input validation, and safeguards that prevent code from running too long. This article walks through five error-handling functions that you can use in tasks like web scraping, application programming interfaces (APIs), processing user data, and more.
You can find the code on GitHub..
# Retrying failed operations with exponential backoff
In many projects, API calls and network requests often fail. A starting approach is to try once and catch any exceptions, log them, and stop. The best way is to try again.
This is where the exponential bake-off comes in. Instead of hammering a failed service with quick retries—which only makes things worse—you wait a little longer between each attempt: 1 second, then 2 seconds, then 4 seconds, and so on.
Let’s create a decorator that does this:
import time
import functools
from typing import Callable, Type, Tuple
def retry_with_backoff(
max_attempts: int = 3,
base_delay: float = 1.0,
exponential_base: float = 2.0,
exceptions: Tuple(Type(Exception), ...) = (Exception,)
):
"""
Retry a function with exponential backoff.
Args:
max_attempts: Maximum number of retry attempts
base_delay: Initial delay in seconds
exponential_base: Multiplier for delay (2.0 = double each time)
exceptions: Tuple of exception types to catch and retry
"""
def decorator(func: Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
delay = base_delay * (exponential_base ** attempt)
print(f"Attempt {attempt + 1} failed: {e}")
print(f"Retrying in {delay:.1f} seconds...")
time.sleep(delay)
else:
print(f"All {max_attempts} attempts failed")
raise last_exception
return wrapper
return decoratorA decorator wraps your function and catches certain exceptions. The key is accounting. delay = base_delay * (exponential_base ** attempt). with the base_delay=1 And exponential_base=2your delays are 1s, 2s, 4s, 8s. This gives stressed systems time to recover.
The exceptions parameter lets you specify which errors to retry. You can try again. ConnectionError But no ValueErrorsince connection problems are temporary but validation errors are not.
Now let’s see it in action:
import random
@retry_with_backoff(max_attempts=4, base_delay=0.5, exceptions=(ConnectionError,))
def fetch_user_data(user_id):
"""Simulate an unreliable API."""
if random.random() < 0.6: # 60% failure rate
raise ConnectionError("Service temporarily unavailable")
return {"id": user_id, "name": "Sara", "status": "active"}
# Watch it retry automatically
result = fetch_user_data(12345)
print(f"Success: {result}")Output:
Success: {'id': 12345, 'name': 'Sara', 'status': 'active'}# Validating input with composable rules
Validating user input is tedious and repetitive. You check if the strings are empty, if the numbers are in range, and if the emails look correct. Before you know it, you’ve got if statements everywhere and your code looks like a mess.
Let’s create an authentication system that is easy to use. First, we need a custom exception:
from typing import Any, Callable, Dict, List, Optional
class ValidationError(Exception):
"""Raised when validation fails."""
def __init__(self, field: str, errors: List(str)):
self.field = field
self.errors = errors
super().__init__(f"{field}: {', '.join(errors)}")This exception has multiple error messages. When validation fails, we want to show the user everything that is wrong, not just the first error.
Now here is the validator:
def validate_input(
value: Any,
field_name: str,
rules: Dict(str, Callable((Any), bool)),
messages: Optional(Dict(str, str)) = None
) -> Any:
"""
Validate input against multiple rules.
Returns the value if valid, raises ValidationError otherwise.
"""
if messages is None:
messages = {}
errors = ()
for rule_name, rule_func in rules.items():
try:
if not rule_func(value):
error_msg = messages.get(
rule_name,
f"Failed validation rule: {rule_name}"
)
errors.append(error_msg)
except Exception as e:
errors.append(f"Validation error in {rule_name}: {str(e)}")
if errors:
raise ValidationError(field_name, errors)
return valuei rules In a dictionary, each rule is simply a function that returns a function. True or False. This makes rules composable and reusable.
Let’s make some general validation rules:
# Reusable validation rules
def not_empty(value: str) -> bool:
return bool(value and value.strip())
def min_length(min_len: int) -> Callable:
return lambda value: len(str(value)) >= min_len
def max_length(max_len: int) -> Callable:
return lambda value: len(str(value)) <= max_len
def in_range(min_val: float, max_val: float) -> Callable:
return lambda value: min_val <= float(value) <= max_valHow to notice min_length, max_lengthand in_range There are factory functions. They return validation functions configured with specified parameters. It lets you write. min_length(3) Instead of creating a new function for each length requirement.
Let’s validate a username:
try:
username = validate_input(
"ab",
"username",
{
"not_empty": not_empty,
"min_length": min_length(3),
"max_length": max_length(20),
},
messages={
"not_empty": "Username cannot be empty",
"min_length": "Username must be at least 3 characters",
"max_length": "Username cannot exceed 20 characters",
}
)
print(f"Valid username: {username}")
except ValidationError as e:
print(f"Invalid: {e}")Output:
Invalid: username: Username must be at least 3 charactersThis approach scales well. Define your rules once, write them as you need them, and get clear error messages.
# Safely navigating nested dictionaries
Access to home dictionaries is often difficult. you get KeyError When the key does not exist, TypeError When you try to subscript a string, and you get cluttered with chains of code. .get() Calls or defenses try-except Working with JavaScript Object Notation (JSON) from blocks APIs makes it more difficult.
Let’s create a function that safely navigates nested structures:
from typing import Any, Optional, List, Union
def safe_get(
data: dict,
path: Union(str, List(str)),
default: Any = None,
separator: str = "."
) -> Any:
"""
Safely get a value from a nested dictionary.
Args:
data: The dictionary to access
path: Dot-separated path (e.g., "user.address.city") or list of keys
default: Value to return if path doesn't exist
separator: Character to split path string (default: ".")
Returns:
The value at the path, or default if not found
"""
# Convert string path to list
if isinstance(path, str):
keys = path.split(separator)
else:
keys = path
current = data
for key in keys:
try:
# Handle list indices (convert string to int if numeric)
if isinstance(current, list):
try:
key = int(key)
except (ValueError, TypeError):
return default
current = current(key)
except (KeyError, IndexError, TypeError):
return default
return currentThe function splits the path into individual keys and navigates the nested structure step by step. If no key exists or if you try to subscribe to an object that isn’t subscribeable, it returns the default instead of crashing.
It also handles list indexes automatically. If the current value is a list and the key is numeric, it converts the key to an integer.
The companion function to set the values is:
def safe_set(
data: dict,
path: Union(str, List(str)),
value: Any,
separator: str = ".",
create_missing: bool = True
) -> bool:
"""
Safely set a value in a nested dictionary.
Args:
data: The dictionary to modify
path: Dot-separated path or list of keys
value: Value to set
separator: Character to split path string
create_missing: Whether to create missing intermediate dicts
Returns:
True if successful, False otherwise
"""
if isinstance(path, str):
keys = path.split(separator)
else:
keys = path
if not keys:
return False
current = data
# Navigate to the parent of the final key
for key in keys(:-1):
if key not in current:
if create_missing:
current(key) = {}
else:
return False
current = current(key)
if not isinstance(current, dict):
return False
# Set the final value
current(keys(-1)) = value
return Truegave safe_set The function creates the nested structure and sets the value as needed. This is useful for dynamically building dictionaries.
Let’s examine both:
# Sample nested data
user_data = {
"user": {
"name": "Anna",
"address": {
"city": "San Francisco",
"zip": "94105"
},
"orders": (
{"id": 1, "total": 99.99},
{"id": 2, "total": 149.50}
)
}
}
# Safe get examples
city = safe_get(user_data, "user.address.city")
print(f"City: {city}")
country = safe_get(user_data, "user.address.country", default="Unknown")
print(f"Country: {country}")
first_order = safe_get(user_data, "user.orders.0.total")
print(f"First order: ${first_order}")
# Safe set example
new_data = {}
safe_set(new_data, "user.settings.theme", "dark")
print(f"Created: {new_data}")Output:
City: San Francisco
Country: Unknown
First order: $99.99
Created: {'user': {'settings': {'theme': 'dark'}}}This pattern eliminates the clutter of defensive programming and cleans up your code when working with JSON, configuration files, or any deeply nested data.
# Enforcing timeout on long operations
Some operations take a long time. A database query can hang, a web scraping operation can get stuck on a slow server, or a calculation can run forever. You need a way to set time limits and bail out.
Here is a timeout decorator using threading:
import threading
import functools
from typing import Callable, Optional
class TimeoutError(Exception):
"""Raised when an operation exceeds its timeout."""
pass
def timeout(seconds: int, error_message: Optional(str) = None):
"""
Decorator to enforce a timeout on function execution.
Args:
seconds: Maximum execution time in seconds
error_message: Custom error message for timeout
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
result = (TimeoutError(
error_message or f"Operation timed out after {seconds} seconds"
))
def target():
try:
result(0) = func(*args, **kwargs)
except Exception as e:
result(0) = e
thread = threading.Thread(target=target)
thread.daemon = True
thread.start()
thread.join(timeout=seconds)
if thread.is_alive():
raise TimeoutError(
error_message or f"Operation timed out after {seconds} seconds"
)
if isinstance(result(0), Exception):
raise result(0)
return result(0)
return wrapper
return decoratorThis decorator runs and uses your function in a separate thread. thread.join(timeout=seconds) Waiting If the thread is still alive after the timeout, we know it took too long and extended TimeoutError.
The result of the function is stored in a list (variable container) so that the inner thread can modify it. If an exception has occurred in a thread, we raise it again in the main thread.
⚠️ A limit: Thread continues to run in background even after timeout. This is fine for most use cases, but for operations with side effects, be careful.
Let’s check it out:
import time
@timeout(2, error_message="Query took too long")
def slow_database_query():
"""Simulate a slow query."""
time.sleep(5)
return "Query result"
@timeout(3)
def fetch_data():
"""Simulate a quick operation."""
time.sleep(1)
return {"data": "value"}
# Test timeout
try:
result = slow_database_query()
print(f"Result: {result}")
except TimeoutError as e:
print(f"Timeout: {e}")
# Test success
try:
data = fetch_data()
print(f"Success: {data}")
except TimeoutError as e:
print(f"Timeout: {e}")Output:
Timeout: Query took too long
Success: {'data': 'value'}This pattern is essential for building responsive applications. When you’re scraping websites, calling external APIs, or running user code, a timeout prevents your program from hanging indefinitely.
# Resource management with automatic cleanup
Files, database connections, and opening network sockets require careful cleaning. If an exception occurs, you must ensure that the resource is released. Context Managers Using the with Statement handles this, but sometimes you need more control.
Let’s create a flexible context manager for automatic resource cleanup:
from contextlib import contextmanager
from typing import Callable, Any, Optional
import traceback
@contextmanager
def managed_resource(
acquire: Callable((), Any),
release: Callable((Any), None),
on_error: Optional(Callable((Exception, Any), None)) = None,
suppress_errors: bool = False
):
"""
Context manager for automatic resource acquisition and cleanup.
Args:
acquire: Function to acquire the resource
release: Function to release the resource
on_error: Optional error handler
suppress_errors: Whether to suppress exceptions after cleanup
"""
resource = None
try:
resource = acquire()
yield resource
except Exception as e:
if on_error and resource is not None:
try:
on_error(e, resource)
except Exception as handler_error:
print(f"Error in error handler: {handler_error}")
if not suppress_errors:
raise
finally:
if resource is not None:
try:
release(resource)
except Exception as cleanup_error:
print(f"Error during cleanup: {cleanup_error}")
traceback.print_exc()gave managed_resource A function is a context manager factory. It takes two required functions: one to acquire the resource and the other to release it. gave release The function runs forever. finally Block guarantees sanity even if exceptions occur.
optional on_error The parameter lets you handle errors before they propagate. This is useful for logging in, sending alerts, or attempting recovery. gave suppress_errors The flag determines whether exceptions are explicitly raised or suppressed.
Here’s a helpful class to demonstrate resource tracking:
class ResourceTracker:
"""Helper class to track resource operations."""
def __init__(self, name: str, verbose: bool = True):
self.name = name
self.verbose = verbose
self.operations = ()
def log(self, operation: str):
self.operations.append(operation)
if self.verbose:
print(f"({self.name}) {operation}")
def acquire(self):
self.log("Acquiring resource")
return self
def release(self):
self.log("Releasing resource")
def use(self, action: str):
self.log(f"Using resource: {action}")Let’s examine the context manager:
# Example: Operation with error handling
tracker = ResourceTracker("Database")
def error_handler(exception, resource):
resource.log(f"Error occurred: {exception}")
resource.log("Attempting rollback")
try:
with managed_resource(
acquire=lambda: tracker.acquire(),
release=lambda r: r.release(),
on_error=error_handler
) as db:
db.use("INSERT INTO users")
raise ValueError("Duplicate entry")
except ValueError as e:
print(f"Caught: {e}")Output:
(Database) Acquiring resource
(Database) Using resource: INSERT INTO users
(Database) Error occurred: Duplicate entry
(Database) Attempting rollback
(Database) Releasing resource
Caught: Duplicate entryThis pattern is useful for managing database connections, file handles, network sockets, locks, and any resources that need to be guaranteed sanity. This prevents resource leaks and secures your code.
# wrap up
Each function in this article addresses a specific error handling challenge: retrying transient failures, systematically validating input, safely accessing nested data, preventing hang operations, and managing resource cleanup.
These patterns appear repeatedly in API integration, data processing pipelines, web scraping, and user-facing applications.
Techniques here use decorators, context managers, and composable functions to make errors less frequent and more reliable. You can leave these functions as is in your projects or adapt them to your specific needs. They are self-contained, easy to understand, and solve problems you will encounter regularly. Happy coding!
Bala Priya c is a developer and technical writer from India. She loves working at the intersection of mathematics, programming, data science, and content creation. His areas of interest and expertise include DevOps, data science, and natural language processing. She enjoys reading, writing, coding and coffee! Currently, she’s working on learning lessons and sharing her knowledge with the developer community, writing tutorials, how-to guides, opinion pieces, and more. Bala also creates engaging resource reviews and coding tutorials.