This guide provides a comprehensive Python implementation for integrating with JUHE API, featuring complete error handling, retries, and best practices.
Prerequisites
- Python 3.6+
requestslibrary for HTTP requests- Basic understanding of Python and asynchronous programming (optional for async examples)
Installation
First, install the required packages:
pip install requests requests-cache backoff
For async support (optional):
pip install aiohttp
Basic Client Implementation
Below is a complete Python client for JUHE API with robust error handling and retries:
import time
import json
import logging
import requests
import backoff
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class JuheAPIClient:
"""JUHE API Client with error handling and retry logic."""
BASE_URL = "<https://hub.juheapi.com>"
def __init__(self, api_key, timeout=10, max_retries=3, backoff_factor=0.5):
"""
Initialize the JUHE API client.
Args:
api_key (str): Your JUHE API key
timeout (int): Request timeout in seconds
max_retries (int): Maximum number of retries for failed requests
backoff_factor (float): Backoff factor for retry delay
"""
self.api_key = api_key
self.timeout = timeout
self.logger = logging.getLogger("juhe_api")
# Configure session with retry logic
self.session = requests.Session()
retry_strategy = Retry(
total=max_retries,
backoff_factor=backoff_factor,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["GET", "POST", "PUT", "DELETE"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
self.session.mount("https://", adapter)
self.session.mount("http://", adapter)
# Default headers
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json"
})
def _get_auth_header(self):
"""Return the authorization header."""
return {"Authorization": f"Bearer {self.api_key}"}
def _handle_response(self, response):
"""
Handle API response and convert errors to exceptions.
Args:
response (requests.Response): API response object
Returns:
dict: Parsed JSON response
Raises:
JuheAPIError: If the API returns an error
"""
try:
data = response.json()
except ValueError:
self.logger.error(f"Invalid JSON response: {response.text}")
raise JuheAPIError("Invalid JSON response from API",
status_code=response.status_code)
# Check for API error response
if response.status_code >= 400 or (data.get("status") == "error"):
error_code = data.get("code", "UNKNOWN_ERROR")
error_message = data.get("message", "Unknown error occurred")
self.logger.error(f"API Error: {error_code} - {error_message}")
raise JuheAPIError(error_message,
status_code=response.status_code,
error_code=error_code)
return data
# Retry decorator for transient errors
@backoff.on_exception(backoff.expo,
(requests.exceptions.RequestException, JuheAPIError),
max_tries=3,
giveup=lambda e: isinstance(e, JuheAPIError) and e.status_code < 500)
def request(self, method, endpoint, params=None, data=None, headers=None):
"""
Make a request to the API with error handling and retries.
Args:
method (str): HTTP method (GET, POST, etc.)
endpoint (str): API endpoint
params (dict, optional): Query parameters
data (dict, optional): Request body data
headers (dict, optional): Additional headers
Returns:
dict: Parsed JSON response
"""
url = f"{self.BASE_URL}{endpoint}"
request_headers = self._get_auth_header()
if headers:
request_headers.update(headers)
# Add API key to params for GET requests
if method.upper() == "GET" and params is None:
params = {}
if params is not None and method.upper() == "GET":
params["key"] = self.api_key
try:
start_time = time.time()
self.logger.debug(f"Making {method} request to {endpoint}")
response = self.session.request(
method=method,
url=url,
params=params,
json=data,
headers=request_headers,
timeout=self.timeout
)
duration = time.time() - start_time
self.logger.debug(f"Request completed in {duration:.2f}s with status {response.status_code}")
return self._handle_response(response)
except requests.exceptions.Timeout:
self.logger.error(f"Request to {endpoint} timed out after {self.timeout}s")
raise JuheAPIError("Request timed out", error_code="TIMEOUT")
except requests.exceptions.ConnectionError:
self.logger.error(f"Connection error while connecting to {endpoint}")
raise JuheAPIError("Connection error", error_code="CONNECTION_ERROR")
except Exception as e:
self.logger.error(f"Unexpected error: {str(e)}")
raise JuheAPIError(f"Unexpected error: {str(e)}", error_code="UNEXPECTED_ERROR")
# Convenience methods for different HTTP methods
def get(self, endpoint, params=None, headers=None):
return self.request("GET", endpoint, params=params, headers=headers)
def post(self, endpoint, data=None, params=None, headers=None):
return self.request("POST", endpoint, params=params, data=data, headers=headers)
def put(self, endpoint, data=None, params=None, headers=None):
return self.request("PUT", endpoint, params=params, data=data, headers=headers)
def delete(self, endpoint, params=None, headers=None):
return self.request("DELETE", endpoint, params=params, headers=headers)
# API-specific methods
def get_weather(self, location):
"""Get current weather for a location."""
return self.get("/weather/current", params={"location": location})
def get_weather_forecast(self, location, days=5):
"""Get weather forecast for a location."""
return self.get("/weather/forecast", params={"location": location, "days": days})
class JuheAPIError(Exception):
"""Custom exception for JUHE API errors."""
def __init__(self, message, status_code=None, error_code=None):
self.message = message
self.status_code = status_code
self.error_code = error_code
super().__init__(self.message)
Using the Client
Here's how to use the client in your application:
import logging
import sys
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler(sys.stdout)]
)
# Initialize the client
client = JuheAPIClient(api_key="YOUR_API_KEY")
try:
# Get current weather
weather = client.get_weather("London")
print(f"Current temperature in London: {weather['data']['current']['temp_c']}°C")
# Get weather forecast
forecast = client.get_weather_forecast("New York", days=3)
print(f"3-day forecast for New York:")
for day in forecast['data']['forecast']:
print(f" {day['date']}: {day['condition']} ({day['min_temp_c']}°C - {day['max_temp_c']}°C)")
except JuheAPIError as e:
print(f"API Error ({e.error_code}): {e.message}")
except Exception as e:
print(f"Unexpected error: {str(e)}")
Advanced Features
Implementing Caching
Add caching to reduce API calls for frequently accessed data:
import requests_cache
# Install with pip install requests_cache
# Initialize a cached session (cache expires after 1 hour)
cached_session = requests_cache.CachedSession(
cache_name='juhe_cache',
backend='sqlite',
expire_after=3600
)
# Modify the client to use the cached session
client = JuheAPIClient(api_key="YOUR_API_KEY")
client.session = cached_session
Batch Processing
Process multiple requests efficiently:
def batch_request(client, requests):
"""
Make multiple API requests in a single batch.
Args:
client (JuheAPIClient): The API client
requests (list): List of request dictionaries
Returns:
dict: Batch response
"""
return client.post("/batch", data={"requests": requests})
# Example usage
batch_results = batch_request(client, [
{
"method": "GET",
"path": "/weather/current",
"params": {"location": "London"}
},
{
"method": "GET",
"path": "/weather/forecast",
"params": {"location": "New York", "days": 3}
}
])
# Process results
for i, result in enumerate(batch_results["results"]):
if result["status"] == "success":
print(f"Request {i+1} succeeded")
else:
print(f"Request {i+1} failed: {result.get('message')}")
Asynchronous Client
For applications that need non-blocking API calls:
import asyncio
import aiohttp
import backoff
from typing import Dict, Any, Optional
class AsyncJuheAPIClient:
"""Asynchronous JUHE API Client."""
BASE_URL = "<https://hub.juheapi.com>"
def __init__(self, api_key: str, timeout: float = 10.0):
self.api_key = api_key
self.timeout = timeout
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={"Content-Type": "application/json"}
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
def _get_auth_header(self):
return {"Authorization": f"Bearer {self.api_key}"}
@backoff.on_exception(backoff.expo,
(aiohttp.ClientError, aiohttp.ClientResponseError),
max_tries=3)
async def request(self, method: str, endpoint: str, **kwargs) -> Dict[str, Any]:
"""Make an async request to the API."""
if self.session is None:
self.session = aiohttp.ClientSession(
headers={"Content-Type": "application/json"}
)
url = f"{self.BASE_URL}{endpoint}"
headers = self._get_auth_header()
if 'headers' in kwargs:
headers.update(kwargs.pop('headers'))
# Add API key to params for GET requests
params = kwargs.get('params', {})
if method.upper() == "GET":
params['key'] = self.api_key
kwargs['params'] = params
try:
async with self.session.request(
method=method,
url=url,
headers=headers,
timeout=self.timeout,
**kwargs
) as response:
data = await response.json()
if response.status >= 400 or data.get('status') == 'error':
error_code = data.get('code', 'UNKNOWN_ERROR')
error_message = data.get('message', 'Unknown error occurred')
raise JuheAPIError(error_message,
status_code=response.status,
error_code=error_code)
return data
except aiohttp.ClientResponseError as e:
raise JuheAPIError(str(e), status_code=e.status)
except aiohttp.ClientError as e:
raise JuheAPIError(str(e))
async def get(self, endpoint: str, **kwargs) -> Dict[str, Any]:
return await self.request("GET", endpoint, **kwargs)
async def post(self, endpoint: str, **kwargs) -> Dict[str, Any]:
return await self.request("POST", endpoint, **kwargs)
async def get_weather(self, location: str) -> Dict[str, Any]:
"""Get current weather for a location asynchronously."""
return await self.get("/weather/current", params={"location": location})
# Example usage
async def main():
async with AsyncJuheAPIClient(api_key="YOUR_API_KEY") as client:
try:
# Make concurrent API calls
weather_task = client.get_weather("London")
forecast_task = client.get("/weather/forecast", params={"location": "Paris", "days": 5})
# Wait for both to complete
weather, forecast = await asyncio.gather(weather_task, forecast_task)
print(f"Current temperature in London: {weather['data']['current']['temp_c']}°C")
print(f"5-day forecast for Paris retrieved successfully")
except JuheAPIError as e:
print(f"API Error ({e.error_code}): {e.message}")
# Run the async example
if __name__ == "__main__":
asyncio.run(main())
Error Handling Best Practices
The client implements comprehensive error handling including:
- Response Validation: Checking for valid JSON and error responses
- Detailed Error Objects: Custom exceptions with status code and error code
- Retry Logic: Automatic retries for transient errors
- Logging: Detailed logs for debugging
- Timeout Management: Configurable timeouts to prevent hanging requests
Complete Project Structure
For larger applications, we recommend organizing your code like this:
juhe_api/
├── __init__.py
├── client.py # Core client implementation
├── exceptions.py # Custom exceptions
├── async_client.py # Async implementation
├── models/ # Response data models
│ ├── __init__.py
│ ├── weather.py
│ └── ...
└── utils/ # Utility functions
├── __init__.py
├── retry.py
└── cache.py
Next Steps
- Check out our Node.js Integration for JavaScript implementations
- Explore Advanced Topics for more optimization techniques
- Visit the API Reference for complete endpoint documentation