7. Complete production dashboard
Five patterns, one application. This page wires the message templates, the categorizer, the retry wrapper, the dual-audience logger, and the input validation into the Chapter 8 Weather Dashboard, and walks through how the pieces connect. The result is the dashboard you started the chapter wanting: one that handles typos, network hiccups, rate limits, and server outages without ever showing a stack trace.
The complete production dashboard
Save this as weather_dashboard.py. Every method references a pattern from an earlier page in this chapter, and the comments point back to where each piece was introduced:
import requests
import time
import random
import logging
import re
from requests.exceptions import Timeout, ConnectionError, HTTPError
from datetime import datetime
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='[%(levelname)s] %(asctime)s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
handlers=[
logging.FileHandler('weather_dashboard.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# Message templates for all error types
MESSAGE_TEMPLATES = {
"user_input": {
"empty": (
"Please enter a city name to get weather information.",
"Type the name of any city or town.",
"Examples: London, Paris, Tokyo"
),
"too_long": (
"That city name is too long (maximum 100 characters).",
"Please enter a shorter name.",
"Examples: London, San Francisco, Mexico City"
),
"invalid_chars": (
"City names can only contain letters, spaces, hyphens, and apostrophes.",
"Please check your input and try again.",
"Examples: London, Saint-Denis, O'Fallon"
)
},
"transient": {
"timeout": (
"We're having trouble connecting to the weather service.",
"This is usually temporary - please try again in a moment.",
"If the problem continues, check your internet connection."
),
"connection": (
"We're having trouble connecting to the weather service.",
"This is usually temporary - please try again in a moment.",
"If the problem continues, check your internet connection."
),
"rate_limit": (
"We've made too many requests to the weather service.",
"Please wait {retry_seconds} seconds before trying again.",
"This helps us stay within service limits."
),
"server_error": (
"The weather service is currently experiencing issues.",
"This is temporary - please try again in a few minutes.",
"If the problem continues, check the service status page."
)
},
"not_found": {
"city": (
"We couldn't find weather data for \"{city_name}\".",
"Please check the spelling or try a nearby city.",
"Examples: {suggestions}"
)
},
"unknown": {
"general": (
"Something unexpected happened while getting weather data.",
"Please try again. If the problem continues, contact support.",
"Error details have been logged for investigation."
)
}
}
def categorize_error(exception, response=None):
"""Categorize any exception into one of four types."""
# User input errors (checked before network calls)
if isinstance(exception, ValueError):
if "empty" in str(exception).lower():
return ("user_input", "empty", {})
elif "too long" in str(exception).lower():
return ("user_input", "too_long", {})
else:
return ("user_input", "invalid_chars", {})
# Transient errors that should be retried
if isinstance(exception, Timeout):
return ("transient", "timeout", {})
if isinstance(exception, ConnectionError):
return ("transient", "connection", {})
if isinstance(exception, HTTPError):
status_code = exception.response.status_code
if status_code == 429:
retry_after = exception.response.headers.get('Retry-After', '60')
return ("transient", "rate_limit", {"retry_seconds": retry_after})
if status_code in [500, 502, 503, 504]:
return ("transient", "server_error", {})
if status_code == 404:
return ("not_found", "city", {})
# API returned empty results (city not found)
if isinstance(exception, KeyError) and exception.args[0] == 'results':
return ("not_found", "city", {})
# Everything else is unknown
return ("unknown", "general", {})
def compose_error_message(category, error_type, **context):
"""Compose a three-part error message from templates."""
template = MESSAGE_TEMPLATES[category][error_type]
# Add default suggestions if missing
if category == "not_found" and "suggestions" not in context:
context["suggestions"] = "London, Paris, Tokyo"
if category == "not_found" and "city_name" not in context:
context["city_name"] = "that city"
# Format each part with context variables
what_happened = template[0].format(**context)
what_to_do = template[1].format(**context)
examples = template[2].format(**context)
return f"{what_happened}\n{what_to_do}\n{examples}"
def log_error(category, error_type, exception, context=None):
"""Log technical details for developers."""
log_entry = {
"timestamp": datetime.now().isoformat(),
"category": category,
"error_type": error_type,
"exception_type": type(exception).__name__,
"exception_message": str(exception),
"context": context or {}
}
if category == "user_input":
logger.info(f"User input rejected: {log_entry}")
elif category == "transient":
logger.warning(f"Transient failure: {log_entry}")
elif category == "not_found":
logger.info(f"Resource not found: {log_entry}")
else: # unknown
logger.error(f"Unknown error: {log_entry}", exc_info=True)
def validate_city_name(city_name):
"""Validate city name before making API calls."""
if not city_name or not city_name.strip():
raise ValueError("City name cannot be empty")
if len(city_name) > 100:
raise ValueError("City name too long (maximum 100 characters)")
if not re.match(r"^[a-zA-Z\s\-']+$", city_name):
raise ValueError("City name contains invalid characters")
return city_name.strip()
def calculate_backoff_with_jitter(attempt, base_delay=1.0):
"""Calculate exponential backoff with jitter."""
exponential_delay = base_delay * (2 ** attempt)
jitter = random.uniform(0, base_delay * 0.5)
return exponential_delay + jitter
def retry_with_backoff(func, *args, max_attempts=3, base_delay=1.0, **kwargs):
"""Retry function with exponential backoff and jitter. Honors Retry-After on 429."""
attempt = 0
while attempt < max_attempts:
try:
result = func(*args, **kwargs)
return result, None
except Exception as e:
category, error_type, context = categorize_error(e)
# Only retry transient errors
if category != "transient":
log_error(category, error_type, e, context)
message = compose_error_message(category, error_type, **context)
return None, message
attempt += 1
if attempt >= max_attempts:
log_error(category, error_type, e, context)
message = compose_error_message(category, error_type,
attempts=attempt, **context)
return None, message
# Honor Retry-After when the server provides one (429 only).
# Otherwise, use exponential backoff with jitter.
if error_type == "rate_limit" and context.get("retry_seconds"):
delay = float(context["retry_seconds"])
else:
delay = calculate_backoff_with_jitter(attempt - 1, base_delay)
logger.warning(
f"Attempt {attempt} failed: {error_type}. "
f"Retrying in {delay:.1f}s..."
)
time.sleep(delay)
return None, "Maximum retry attempts exceeded"
def find_location(city_name):
"""Find location coordinates for a city with error handling."""
try:
validated_name = validate_city_name(city_name)
except ValueError as e:
category, error_type, context = categorize_error(e)
log_error(category, error_type, e, {"user_input": city_name})
message = compose_error_message(category, error_type, city_name=city_name)
return None, message
def make_location_request():
response = requests.get(
"https://geocoding-api.open-meteo.com/v1/search",
params={"name": validated_name, "count": 1},
timeout=5
)
response.raise_for_status()
data = response.json()
if not data.get("results"):
raise KeyError("results")
location = data["results"][0]
return {
"lat": location["latitude"],
"lon": location["longitude"],
"name": location["name"]
}
result, error = retry_with_backoff(make_location_request)
if error:
# Replace the fallback label with the actual input this call knows about.
error = error.replace("that city", city_name)
return None, error
return result, None
def get_weather(location):
"""Get weather data for location coordinates with error handling."""
def make_weather_request():
response = requests.get(
"https://api.open-meteo.com/v1/forecast",
params={
"latitude": location["lat"],
"longitude": location["lon"],
"current_weather": True
},
timeout=5
)
response.raise_for_status()
data = response.json()
current = data["current_weather"]
return {
"temperature": current["temperature"],
"windspeed": current["windspeed"],
"conditions": "Clear" if current["weathercode"] == 0 else "Cloudy"
}
result, error = retry_with_backoff(make_weather_request)
return result, error
class WeatherDashboard:
"""Production weather dashboard with complete error handling."""
def get_weather_for_city(self, city_name):
"""Get weather for a city with complete error handling."""
print(f"\nLooking up coordinates for '{city_name}'...")
location, error = find_location(city_name)
if error:
return None, error
print(f"Found: {location['name']}")
print(f"\nFetching weather data...")
weather, error = get_weather(location)
if error:
return None, error
return weather, None
def run(self):
"""Run the interactive dashboard."""
print("=" * 50)
print("Weather Dashboard (Production)")
print("=" * 50)
while True:
city_name = input("\nEnter city name (or 'quit' to exit): ").strip()
if city_name.lower() in ['quit', 'exit', 'q']:
print("Goodbye!")
break
weather, error = self.get_weather_for_city(city_name)
if error:
print(f"\n{error}")
else:
print(f"\n{'=' * 50}")
print(f"Weather for {city_name}")
print(f"{'=' * 50}")
print(f"Temperature: {weather['temperature']}°C")
print(f"Wind Speed: {weather['windspeed']} km/h")
print(f"Conditions: {weather['conditions']}")
print(f"{'=' * 50}")
if __name__ == "__main__":
dashboard = WeatherDashboard()
dashboard.run()This production implementation integrates all five concepts:
- User-Centric Communication: Three-part error messages from centralized templates
- Systematic Categorization: All errors map to four categories with consistent handling
- Production-Grade Recovery: Automatic retry with exponential backoff and jitter for transient failures
- Dual-Audience Logging: Friendly messages for users, technical logs for developers
- Automated Testing: Complete test suite verifies all error handling works correctly
How the pieces connect
Let's trace through what happens when an error occurs in the production system:
Categorization drives everything: it determined this wasn't retryable, selected the right message template, logged appropriately, and showed helpful user guidance. The system handled the error gracefully without exposing technical details or making pointless retry attempts.
Where this sits relative to APIClient
The weather_dashboard.py above is the single-file version: every pattern this chapter built, composed inline, talking to requests directly so you can see the exception flow end to end. That's the right shape when the exceptions from the underlying library are what you want to categorise, and it's the shape a CLI tool or a script will usually take.
Chapter 7's APIClient makes a different choice. Its _make_request catches exceptions internally and returns a (success, data, message) tuple; the caller never sees the underlying Timeout or HTTPError. Ch9's categorize_error is the opposite: it expects an exception to inspect with isinstance. Those two contracts don't compose cleanly. Making them work together means string-matching against APIClient's internal message text ("Rate limited", "Request timed out", "Authentication failed"), which is exactly the fragile pattern the rest of this chapter was built to avoid.
The honest design choice is to pick one contract per client. If you want the dashboard to handle errors as exceptions so the categoriser can read their types directly, you do what this page does — call requests yourself and let the exceptions propagate. If you want a reusable base class that absorbs credential validation, retries, and 429 handling behind one method, you subclass APIClient and write your categoriser against its tuple shape instead. Both are reasonable; the mistake is trying to run them through each other and ending up matching on strings that were never meant to be an interface.
The patterns on this page are the building blocks either way. The retry loop, the backoff math, the jitter, the categorisation, the three-part message composer, the dual-audience logger — those don't change. What changes is whether you apply them against raw exceptions (this chapter) or against tuple returns from a base class that already handled the HTTP-level details (Chapter 7's shape).