import json
import time
import requests
from typing import Dict, Any
def collect_payload_metrics(url: str, max_attempts: int = 3, timeout: int = 5) -> Dict[str, Any]:
"""
Collects metrics of JSON payloads from a given URL with a timeout and retry mechanism.
Args:
url: The URL to fetch the JSON payload from.
max_attempts: Maximum number of retry attempts.
timeout: Timeout in seconds for each request.
Returns:
A dictionary containing metrics such as success/failure, response time,
and potential error messages.
"""
metrics = {
"success": False,
"response_time": None,
"error_message": None,
"payload_size": None
}
for attempt in range(max_attempts):
try:
start_time = time.time()
response = requests.get(url, timeout=timeout)
end_time = time.time()
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
metrics["success"] = True
metrics["response_time"] = end_time - start_time
metrics["payload_size"] = len(response.content)
try:
payload = response.json()
except json.JSONDecodeError:
metrics["error_message"] = "Invalid JSON response"
break
return metrics # Exit on success
except requests.exceptions.RequestException as e:
metrics["error_message"] = str(e)
if attempt < max_attempts - 1:
print(f"Attempt {attempt + 1} failed: {e}. Retrying...")
time.sleep(1) # Wait before retrying
else:
print(f"Max attempts reached. Failed to retrieve payload.")
return metrics
except Exception as e:
metrics["error_message"] = str(e)
return metrics # Return on unexpected error
if __name__ == '__main__':
# Example usage
test_url = "https://jsonplaceholder.typicode.com/todos/1" # Replace with your URL
metrics = collect_payload_metrics(test_url)
print(metrics)
test_url_invalid = "https://httpstat.us/500" #Example of a failure
metrics = collect_payload_metrics(test_url_invalid)
print(metrics)
Add your comment