Chapter 12: API Data Validation

{
  "current": {
    "temperature_2m": 22.5,
    "relative_humidity_2m": 65,
    "wind_speed_10m": 12.3,
    "weather_code": 0,
    "apparent_temperature": 21.8
  },
  "current_units": {
    "temperature_2m": "°C",
    "relative_humidity_2m": "%"
  }
}

{
  "current": {
    "temperature_2m": -999,         // Sensor error placeholder
    "relative_humidity_2m": "N/A",  // String instead of number
    "wind_speed_10m": null,         // Missing data
    "weather_code": 71,             // Snow code
    "apparent_temperature": 150     // Impossible value
  }
}

def display_current_weather(weather_data, location_name):
    """Display current weather - crashes on bad data."""
    current = weather_data["current"]
    
    # Crashes if "current" missing (KeyError)
    temp = current["temperature_2m"]
    print(f"Temperature: {temp}°C")
    
    # Crashes if temperature is string (TypeError)
    if temp > 30:
        print("It's hot today!")
    
    # Shows nonsense if temp is -999 or 150
    # Users see: "Temperature: -999°C" or "Temperature: 150°C"

def validate_temperature(value):
    """Validate temperature field."""
    if value is None:
        return False, "Temperature missing"
    try:
        temp = float(value)
        if temp < -100 or temp > 60:
            return False, f"Temperature out of range: {temp}"
    except (ValueError, TypeError):
        return False, f"Temperature not numeric: {value}"
    return True, None

def validate_humidity(value):
    """Validate humidity field."""
    if value is None:
        return False, "Humidity missing"
    try:
        humidity = float(value)
        if humidity < 0 or humidity > 100:
            return False, f"Humidity out of range: {humidity}"
    except (ValueError, TypeError):
        return False, f"Humidity not numeric: {value}"
    return True, None

def validate_wind_speed(value):
    """Validate wind speed field."""
    if value is None:
        return False, "Wind speed missing"
    try:
        wind = float(value)
        if wind < 0 or wind > 200:
            return False, f"Wind speed out of range: {wind}"
    except (ValueError, TypeError):
        return False, f"Wind speed not numeric: {value}"
    return True, None

# You need similar functions for:
# - pressure, visibility, precipitation
# - cloud_cover, uv_index, wind_direction
# - and 10 more fields...
#
# That's 50+ lines per field × 15 fields = 750+ lines
# Every API change means updating multiple validators
# New developers must learn your pattern for every field type

def validate_weather_structure(data):
    """
    Layer 1: Validate weather response structure.
    Returns (is_valid, error_message).
    """
    # Root must be a dictionary
    if not isinstance(data, dict):
        return False, "Weather data must be a dictionary"
    
    # Must contain 'current' section
    if "current" not in data:
        return False, "Weather data missing 'current' section"
    
    # 'current' must be a dictionary
    if not isinstance(data["current"], dict):
        return False, "'current' section must be a dictionary"
    
    # Check for current_units (helpful but not critical)
    if "current_units" not in data:
        print("Warning: Missing units information")
    
    return True, None

def validate_weather_content(current_data):
    """
    Layer 2: Validate weather content for realistic values.
    Returns (is_valid, error_message).
    """
    # Temperature is required
    if "temperature_2m" not in current_data:
        return False, "Missing required temperature data"
    
    temp = current_data["temperature_2m"]
    if temp is None:
        return False, "Temperature cannot be null"
    
    # Validate temperature is numeric and realistic
    try:
        temp_float = float(temp)
        if temp_float < -100 or temp_float > 60:
            return False, f"Unrealistic temperature: {temp_float}°C"
    except (ValueError, TypeError):
        return False, f"Temperature must be numeric, got: {temp}"
    
    # Optional field: humidity (if present, must be valid)
    if "relative_humidity_2m" in current_data:
        humidity = current_data["relative_humidity_2m"]
        if humidity is not None:
            try:
                humidity_float = float(humidity)
                if humidity_float < 0 or humidity_float > 100:
                    return False, f"Invalid humidity: {humidity_float}%"
            except (ValueError, TypeError):
                return False, f"Humidity must be numeric, got: {humidity}"
    
    # Optional field: wind speed (if present, must be valid)
    if "wind_speed_10m" in current_data:
        wind = current_data["wind_speed_10m"]
        if wind is not None:
            try:
                wind_float = float(wind)
                if wind_float < 0:
                    return False, f"Wind speed cannot be negative: {wind_float}"
                if wind_float > 200:
                    return False, f"Unrealistic wind speed: {wind_float} km/h"
            except (ValueError, TypeError):
                return False, f"Wind speed must be numeric, got: {wind}"
    
    return True, None

def validate_weather_business_rules(current_data):
    """
    Layer 3: Validate weather business rules and logical consistency.
    Returns (is_valid, error_message).
    """
    temp = current_data.get("temperature_2m")
    weather_code = current_data.get("weather_code")
    
    # Business rule: Snow conditions should match temperature
    if temp is not None and weather_code is not None:
        try:
            temp_float = float(temp)
            code_int = int(weather_code)
            
            # Weather codes 71, 73, 75 indicate snow
            if code_int in [71, 73, 75] and temp_float > 5:
                return False, f"Snow conditions at {temp_float}°C is unlikely"
                
        except (ValueError, TypeError):
            # Type errors already caught in content validation
            pass
    
    # Business rule: "Feels like" should be near actual temperature
    apparent_temp = current_data.get("apparent_temperature")
    if temp is not None and apparent_temp is not None:
        try:
            temp_float = float(temp)
            apparent_float = float(apparent_temp)
            
            # "Feels like" shouldn't differ by more than 20°C
            if abs(apparent_float - temp_float) > 20:
                return False, (
                    f"'Feels like' {apparent_float}°C too different "
                    f"from actual {temp_float}°C"
                )
        except (ValueError, TypeError):
            # Type errors already caught in content validation
            pass
    
    return True, None

def validate_weather_data(data):
    """
    Complete validation pipeline: structure → content → business rules.
    Returns (is_valid, error_message).
    """
    # Layer 1: Structural validation
    valid, error = validate_weather_structure(data)
    if not valid:
        return False, f"Structure validation failed: {error}"
    
    # Layer 2: Content validation
    current = data["current"]  # Safe because structure validated
    valid, error = validate_weather_content(current)
    if not valid:
        return False, f"Content validation failed: {error}"
    
    # Layer 3: Business rules validation
    valid, error = validate_weather_business_rules(current)
    if not valid:
        return False, f"Business rule validation failed: {error}"
    
    # All three layers passed
    return True, None


# Test with valid data
good_data = {
    "current": {
        "temperature_2m": 22.5,
        "relative_humidity_2m": 65,
        "wind_speed_10m": 12.3,
        "weather_code": 0,
        "apparent_temperature": 21.8
    }
}

valid, error = validate_weather_data(good_data)
print(f"Valid: {valid}")  # Valid: True


# Test with bad data
bad_data = {
    "current": {
        "temperature_2m": -999,  # Placeholder value
        "relative_humidity_2m": 65
    }
}

valid, error = validate_weather_data(bad_data)
print(f"Valid: {valid}")
print(f"Error: {error}")
# Valid: False
# Error: Content validation failed: Unrealistic temperature: -999°C

# Running validate_weather_data(bad_data)

Layer 1 — Structural validation  → PASS
Layer 2 — Content validation     → FAIL
Layer 3 — Business rules         → SKIPPED

# Valid: False
# Error: Content validation failed: Unrealistic temperature: -999°C

import requests

class ValidatedWeatherDashboard:
    """Weather dashboard with three-layer validation."""
    
    def __init__(self):
        self.weather_url = "https://api.open-meteo.com/v1/forecast"
    
    def get_weather_data(self, latitude, longitude):
        """Fetch and validate weather data."""
        print("Fetching weather data...")
        
        params = {
            "latitude": latitude,
            "longitude": longitude,
            "current": ["temperature_2m", "relative_humidity_2m", 
                       "wind_speed_10m", "weather_code", "apparent_temperature"],
            "timezone": "auto"
        }
        
        try:
            # Fetch data (error handling from Chapter 9)
            response = requests.get(self.weather_url, params=params, timeout=15)
            response.raise_for_status()
            data = response.json()
            
            # Validate data (new validation pipeline)
            valid, error = validate_weather_data(data)
            if not valid:
                print(f"\n❌ Data validation failed: {error}")
                return None
            
            print("✓ Data validated successfully")
            return data
            
        except requests.exceptions.RequestException as e:
            print(f"\n❌ Request failed: {e}")
            return None
    
    def display_weather(self, data):
        """Display validated weather data."""
        if not data:
            return
        
        current = data["current"]
        
        # After validation passes, we can trust this data
        temp = current["temperature_2m"]
        humidity = current.get("relative_humidity_2m", "N/A")
        wind = current.get("wind_speed_10m", "N/A")
        
        print(f"\n🌡️  Temperature: {temp}°C")
        print(f"💧 Humidity: {humidity}%")
        print(f"💨 Wind Speed: {wind} km/h")


# Usage
dashboard = ValidatedWeatherDashboard()

# Dublin coordinates
weather_data = dashboard.get_weather_data(53.3498, -6.2603)
dashboard.display_weather(weather_data)

{
  "type": "object",
  "required": ["current"],
  "properties": {
    "current": {
      "type": "object",
      "required": ["temperature_2m"],
      "properties": {
        "temperature_2m": {
          "type": "number",
          "minimum": -100,
          "maximum": 60
        },
        "relative_humidity_2m": {
          "type": "number",
          "minimum": 0,
          "maximum": 100
        },
        "wind_speed_10m": {
          "type": "number",
          "minimum": 0,
          "maximum": 200
        },
        "weather_code": {
          "type": "integer"
        },
        "apparent_temperature": {
          "type": "number"
        }
      }
    },
    "current_units": {
      "type": "object"
    }
  }
}

from jsonschema import validate, ValidationError

# Define schema (in practice, load from a JSON file)
weather_schema = {
    "type": "object",
    "required": ["current"],
    "properties": {
        "current": {
            "type": "object",
            "required": ["temperature_2m"],
            "properties": {
                "temperature_2m": {
                    "type": "number",
                    "minimum": -100,
                    "maximum": 60
                },
                "relative_humidity_2m": {
                    "type": "number",
                    "minimum": 0,
                    "maximum": 100
                },
                "wind_speed_10m": {
                    "type": "number",
                    "minimum": 0,
                    "maximum": 200
                }
            }
        }
    }
}

def validate_weather_with_schema(data):
    """Validate weather data using JSON Schema."""
    try:
        validate(instance=data, schema=weather_schema)
        return True, None
    except ValidationError as e:
        return False, e.message


# Test with valid data
good_data = {
    "current": {
        "temperature_2m": 22.5,
        "relative_humidity_2m": 65,
        "wind_speed_10m": 12.3
    }
}

valid, error = validate_weather_with_schema(good_data)
print(f"Valid: {valid}")  # Valid: True


# Test with invalid data
bad_data = {
    "current": {
        "temperature_2m": 150  # Too hot!
    }
}

valid, error = validate_weather_with_schema(bad_data)
print(f"Valid: {valid}")
print(f"Error: {error}")
# Valid: False
# Error: 150 is greater than the maximum of 60

from jsonschema import validate, ValidationError

# Schema handles structure and content
weather_schema = {
    "type": "object",
    "required": ["current"],
    "properties": {
        "current": {
            "type": "object",
            "required": ["temperature_2m"],
            "properties": {
                "temperature_2m": {
                    "type": "number",
                    "minimum": -100,
                    "maximum": 60
                },
                "relative_humidity_2m": {
                    "type": "number",
                    "minimum": 0,
                    "maximum": 100
                },
                "wind_speed_10m": {
                    "type": "number",
                    "minimum": 0,
                    "maximum": 200
                },
                "weather_code": {"type": "integer"},
                "apparent_temperature": {"type": "number"}
            }
        }
    }
}

def validate_business_rules(current_data):
    """Manual validation for business rules only."""
    temp = current_data.get("temperature_2m")
    weather_code = current_data.get("weather_code")
    apparent_temp = current_data.get("apparent_temperature")
    
    # Business rule: Snow at warm temperatures
    if temp is not None and weather_code is not None:
        if weather_code in [71, 73, 75] and temp > 5:
            return False, f"Snow at {temp}°C is unlikely"
    
    # Business rule: "Feels like" vs actual temperature
    if temp is not None and apparent_temp is not None:
        if abs(apparent_temp - temp) > 20:
            return False, (
                f"'Feels like' {apparent_temp}°C too different "
                f"from actual {temp}°C"
            )
    
    return True, None


def validate_weather_hybrid(data):
    """
    Hybrid approach: Schema for structure/content, manual for business rules.
    Returns (is_valid, error_message).
    """
    # Step 1: Schema validation (automated)
    try:
        validate(instance=data, schema=weather_schema)
    except ValidationError as e:
        return False, f"Schema validation failed: {e.message}"
    
    # Step 2: Business rules (manual)
    current = data["current"]
    valid, error = validate_business_rules(current)
    if not valid:
        return False, f"Business rule validation failed: {error}"
    
    return True, None


# Test with valid data
good_data = {
    "current": {
        "temperature_2m": 22.5,
        "relative_humidity_2m": 65,
        "wind_speed_10m": 12.3,
        "weather_code": 0,
        "apparent_temperature": 21.8
    }
}

valid, error = validate_weather_hybrid(good_data)
print(f"Valid: {valid}")  # Valid: True


# Test with business rule violation
bad_data = {
    "current": {
        "temperature_2m": 15.0,  # Warm temperature
        "weather_code": 71  # Snow code - doesn't make sense!
    }
}

valid, error = validate_weather_hybrid(bad_data)
print(f"Valid: {valid}")
print(f"Error: {error}")
# Valid: False
# Error: Business rule validation failed: Snow at 15.0°C is unlikely

def get_weather_with_validation(latitude, longitude):
    """Fetch weather with appropriate failure handling."""
    data = fetch_weather_data(latitude, longitude)
    
    # Fail-fast for critical data
    valid, error = validate_temperature(data["current"]["temperature_2m"])
    if not valid:
        raise ValidationError(f"Critical field invalid: {error}")
    
    # Fail-graceful for enhancements
    valid, error = validate_weather_icon(data["current"].get("weather_code"))
    if not valid:
        logger.warning(f"Optional field invalid: {error}")
        data["current"]["weather_code"] = None  # Use fallback icon
    
    return data

def test_valid_data_passes():
    """Valid weather data should pass all validation layers."""
    data = {
        "current": {
            "temperature_2m": 22.5,
            "relative_humidity_2m": 65,
            "wind_speed_10m": 12.3,
            "weather_code": 0
        }
    }
    
    valid, error = validate_weather_data(data)
    assert valid is True
    assert error is None


def test_missing_section_fails():
    """Missing 'current' section should fail structural validation."""
    data = {"other_section": {}}
    
    valid, error = validate_weather_data(data)
    assert valid is False
    assert "current" in error.lower()


def test_invalid_temperature_fails():
    """Out-of-range temperature should fail content validation."""
    data = {
        "current": {
            "temperature_2m": 150  # Too hot
        }
    }
    
    valid, error = validate_weather_data(data)
    assert valid is False
    assert "temperature" in error.lower()


def test_business_rule_violation_fails():
    """Snow at warm temperature should fail business rules."""
    data = {
        "current": {
            "temperature_2m": 15,
            "weather_code": 71  # Snow code
        }
    }
    
    valid, error = validate_weather_data(data)
    assert valid is False
    assert "snow" in error.lower()


def test_edge_cases():
    """Edge case values should be handled correctly."""
    # Exact boundary values
    data = {"current": {"temperature_2m": -100}}  # Minimum valid
    valid, _ = validate_weather_data(data)
    assert valid is True
    
    data = {"current": {"temperature_2m": 60}}  # Maximum valid
    valid, _ = validate_weather_data(data)
    assert valid is True
    
    # Just outside boundaries
    data = {"current": {"temperature_2m": -100.1}}
    valid, error = validate_weather_data(data)
    assert valid is False

{
  "user": {
    "id": 12345,
    "email": "user@example.com",
    "age": 28,
    "username": "john_doe",
    "preferences": {
      "newsletter": true,
      "theme": "dark"
    }
  }
}

Requirements:

• Structure: Root must have "user" object with id, email, age, username
• Content: Email must contain "@", age between 13-120, username 3-20 characters
• Business rules: If newsletter is true, email cannot be from disposable domains (tempmail, guerrillamail)

Task: Build validate_user_profile(data) following the three-layer pattern.

def validate_user_profile_structure(data):
    """Layer 1: Structural validation."""
    if not isinstance(data, dict):
        return False, "Data must be a dictionary"
    
    if "user" not in data:
        return False, "Missing 'user' object"
    
    user = data["user"]
    if not isinstance(user, dict):
        return False, "'user' must be a dictionary"
    
    required = ["id", "email", "age", "username"]
    for field in required:
        if field not in user:
            return False, f"Missing required field: {field}"
    
    return True, None


def validate_user_profile_content(user):
    """Layer 2: Content validation."""
    # Email format
    email = user["email"]
    if "@" not in email:
        return False, "Email must contain '@'"
    
    # Age range
    try:
        age = int(user["age"])
        if age < 13 or age > 120:
            return False, f"Age {age} outside valid range (13-120)"
    except (ValueError, TypeError):
        return False, f"Age must be integer, got: {user['age']}"
    
    # Username length
    username = user["username"]
    if not isinstance(username, str):
        return False, "Username must be string"
    if len(username) < 3 or len(username) > 20:
        return False, f"Username length {len(username)} outside valid range (3-20)"
    
    return True, None


def validate_user_profile_business_rules(user):
    """Layer 3: Business rules."""
    disposable_domains = ["tempmail.com", "guerrillamail.com"]
    
    preferences = user.get("preferences", {})
    if preferences.get("newsletter"):
        email = user["email"]
        domain = email.split("@")[-1]
        if domain in disposable_domains:
            return False, f"Newsletter requires non-disposable email (got {domain})"
    
    return True, None


def validate_user_profile(data):
    """Complete validation pipeline."""
    valid, error = validate_user_profile_structure(data)
    if not valid:
        return False, f"Structure: {error}"
    
    user = data["user"]
    
    valid, error = validate_user_profile_content(user)
    if not valid:
        return False, f"Content: {error}"
    
    valid, error = validate_user_profile_business_rules(user)
    if not valid:
        return False, f"Business rule: {error}"
    
    return True, None


# Test
test_data = {
    "user": {
        "id": 12345,
        "email": "user@example.com",
        "age": 28,
        "username": "john_doe",
        "preferences": {"newsletter": True, "theme": "dark"}
    }
}

valid, error = validate_user_profile(test_data)
print(f"Valid: {valid}, Error: {error}")