Exceptions
Custom exception classes for handling CIMIS API errors.
Exception Hierarchy
All CIMIS-specific exceptions inherit from the base CimisAPIError class:
CimisAPIError
├── CimisAuthenticationError
└── CimisConnectionError
CimisAPIError
Base exception for all CIMIS API errors.
- class python_cimis.exceptions.CimisAPIError(message, error_code=None, http_code=None)[source]
Bases:
CimisErrorException raised when the CIMIS API returns an error response.
Usage Examples
from python_cimis.exceptions import CimisAPIError
try:
weather_data = client.get_daily_data(
targets=[2],
start_date="2023-06-01",
end_date="2023-06-07"
)
except CimisAPIError as e:
print(f"API Error: {e.message}")
if e.error_code:
print(f"Error Code: {e.error_code}")
if e.response:
print(f"HTTP Status: {e.response.status_code}")
CimisAuthenticationError
Exception raised for authentication-related errors.
- class python_cimis.exceptions.CimisAuthenticationError(message, error_code=None, http_code=None)[source]
Bases:
CimisAPIErrorException raised when there are authentication issues with the API key.
Common Causes
Invalid API key
Expired API key
API key not provided
Account suspended or deactivated
Usage Examples
from python_cimis.exceptions import CimisAuthenticationError
try:
client = CimisClient(app_key="invalid-key")
weather_data = client.get_daily_data(targets=[2], start_date="2023-06-01", end_date="2023-06-07")
except CimisAuthenticationError as e:
print(f"Authentication failed: {e.message}")
print("Please check your API key and ensure it's valid")
CimisConnectionError
Exception raised for connection-related errors.
- class python_cimis.exceptions.CimisConnectionError[source]
Bases:
CimisErrorException raised when there are connection issues with the CIMIS API.
Common Causes
Network connectivity issues
CIMIS server temporarily unavailable
Request timeout
DNS resolution problems
Firewall or proxy blocking requests
Usage Examples
from python_cimis.exceptions import CimisConnectionError
import time
def robust_request(client, max_retries=3):
"""Request with retry logic for connection errors."""
for attempt in range(max_retries):
try:
return client.get_daily_data(
targets=[2],
start_date="2023-06-01",
end_date="2023-06-07"
)
except CimisConnectionError as e:
print(f"Connection error (attempt {attempt + 1}): {e.message}")
if attempt < max_retries - 1:
# Exponential backoff
wait_time = 2 ** attempt
print(f"Retrying in {wait_time} seconds...")
time.sleep(wait_time)
else:
print("Max retries exceeded")
raise
Error Handling Patterns
Basic Error Handling
from python_cimis.exceptions import CimisAPIError, CimisConnectionError, CimisAuthenticationError
def safe_api_call(client, **kwargs):
"""Safely call the CIMIS API with error handling."""
try:
return client.get_daily_data(**kwargs)
except CimisAuthenticationError as e:
print(f"❌ Authentication Error: {e.message}")
print("💡 Solution: Check your API key")
return None
except CimisConnectionError as e:
print(f"🌐 Connection Error: {e.message}")
print("💡 Solution: Check your internet connection and try again")
return None
except CimisAPIError as e:
print(f"⚠️ API Error: {e.message}")
if e.error_code:
print(f"📋 Error Code: {e.error_code}")
return None
Advanced Error Handling
import logging
from typing import Optional
from python_cimis import CimisClient, WeatherData
from python_cimis.exceptions import CimisAPIError, CimisConnectionError, CimisAuthenticationError
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class RobustCimisClient:
"""A wrapper around CimisClient with robust error handling."""
def __init__(self, app_key: str, max_retries: int = 3, timeout: int = 30):
self.client = CimisClient(app_key=app_key, timeout=timeout)
self.max_retries = max_retries
def get_daily_data_robust(self, **kwargs) -> Optional[WeatherData]:
"""Get daily data with comprehensive error handling and retries."""
for attempt in range(self.max_retries):
try:
logger.info(f"Attempting API call (attempt {attempt + 1})")
result = self.client.get_daily_data(**kwargs)
logger.info("API call successful")
return result
except CimisAuthenticationError as e:
logger.error(f"Authentication error: {e.message}")
# Don't retry authentication errors
break
except CimisConnectionError as e:
logger.warning(f"Connection error on attempt {attempt + 1}: {e.message}")
if attempt < self.max_retries - 1:
wait_time = 2 ** attempt
logger.info(f"Retrying in {wait_time} seconds...")
time.sleep(wait_time)
else:
logger.error("Max retries exceeded for connection error")
except CimisAPIError as e:
logger.error(f"API error: {e.message} (Code: {e.error_code})")
# Don't retry generic API errors
break
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
break
return None
Specific Error Scenarios
Invalid API Key
def check_api_key(client):
"""Verify that the API key is valid."""
try:
# Try a simple request to verify the key
stations = client.get_stations()
print("✅ API key is valid")
return True
except CimisAuthenticationError:
print("❌ Invalid API key")
print("Please check:")
print(" 1. API key is correct")
print(" 2. API key is not expired")
print(" 3. Account is active")
return False
Network Issues
import requests
from python_cimis.exceptions import CimisConnectionError
def diagnose_connection_issue():
"""Diagnose connection issues."""
try:
# Test basic connectivity to CIMIS
response = requests.get("https://et.water.ca.gov", timeout=10)
if response.status_code == 200:
print("✅ Can reach CIMIS website")
else:
print(f"⚠️ CIMIS website returned status {response.status_code}")
except requests.ConnectionError:
print("❌ Cannot connect to CIMIS website")
print("Check your internet connection")
except requests.Timeout:
print("⏱️ Connection to CIMIS website timed out")
print("Try increasing the timeout or check network speed")
Rate Limiting
import time
from datetime import datetime
class RateLimitedClient:
"""Client with built-in rate limiting."""
def __init__(self, app_key, requests_per_minute=60):
self.client = CimisClient(app_key=app_key)
self.requests_per_minute = requests_per_minute
self.request_times = []
def _wait_if_needed(self):
"""Wait if rate limit would be exceeded."""
now = datetime.now()
# Remove requests older than 1 minute
cutoff = now.timestamp() - 60
self.request_times = [t for t in self.request_times if t > cutoff]
# Wait if we're at the limit
if len(self.request_times) >= self.requests_per_minute:
wait_time = 60 - (now.timestamp() - self.request_times[0])
if wait_time > 0:
print(f"Rate limiting: waiting {wait_time:.1f} seconds")
time.sleep(wait_time)
self.request_times.append(now.timestamp())
def get_daily_data(self, **kwargs):
"""Get daily data with rate limiting."""
self._wait_if_needed()
return self.client.get_daily_data(**kwargs)
Logging and Monitoring
Setting Up Logging
import logging
from python_cimis.exceptions import CimisAPIError
# Configure logging for CIMIS operations
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('cimis_client.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger('cimis_client')
def logged_api_call(client, operation_name, **kwargs):
"""Make API call with comprehensive logging."""
logger.info(f"Starting {operation_name} with parameters: {kwargs}")
try:
result = client.get_daily_data(**kwargs)
logger.info(f"{operation_name} completed successfully")
return result
except CimisAPIError as e:
logger.error(f"{operation_name} failed: {e.message}")
if e.error_code:
logger.error(f"Error code: {e.error_code}")
raise
Error Reporting
def create_error_report(exception, context=None):
"""Create a detailed error report for troubleshooting."""
import traceback
from datetime import datetime
report = {
'timestamp': datetime.now().isoformat(),
'exception_type': type(exception).__name__,
'message': str(exception),
'context': context or {}
}
if isinstance(exception, CimisAPIError):
report['error_code'] = getattr(exception, 'error_code', None)
report['response_status'] = getattr(exception.response, 'status_code', None) if hasattr(exception, 'response') and exception.response else None
report['traceback'] = traceback.format_exc()
return report
# Usage
try:
weather_data = client.get_daily_data(targets=[2], start_date="2023-06-01", end_date="2023-06-07")
except CimisAPIError as e:
error_report = create_error_report(e, {'targets': [2], 'date_range': '2023-06-01 to 2023-06-07'})
logger.error(f"Error report: {error_report}")
Best Practices
Always use specific exception types for different error scenarios
Implement retry logic for connection errors but not for authentication errors
Log errors appropriately for debugging and monitoring
Provide helpful error messages to users
Validate inputs before making API calls to prevent errors
Use timeouts to prevent hanging requests
Monitor API usage to stay within rate limits
Testing Error Scenarios
import pytest
from python_cimis.exceptions import CimisAuthenticationError, CimisConnectionError
def test_invalid_api_key():
"""Test handling of invalid API key."""
client = CimisClient(app_key="invalid-key")
with pytest.raises(CimisAuthenticationError):
client.get_daily_data(targets=[2], start_date="2023-06-01", end_date="2023-06-07")
def test_connection_error_handling():
"""Test connection error handling."""
# Mock network issues and test retry logic
# Implementation depends on your testing framework