1. import json
  2. import time
  3. import requests
  4. from typing import Dict, Any
  5. def collect_payload_metrics(url: str, max_attempts: int = 3, timeout: int = 5) -> Dict[str, Any]:
  6. """
  7. Collects metrics of JSON payloads from a given URL with a timeout and retry mechanism.
  8. Args:
  9. url: The URL to fetch the JSON payload from.
  10. max_attempts: Maximum number of retry attempts.
  11. timeout: Timeout in seconds for each request.
  12. Returns:
  13. A dictionary containing metrics such as success/failure, response time,
  14. and potential error messages.
  15. """
  16. metrics = {
  17. "success": False,
  18. "response_time": None,
  19. "error_message": None,
  20. "payload_size": None
  21. }
  22. for attempt in range(max_attempts):
  23. try:
  24. start_time = time.time()
  25. response = requests.get(url, timeout=timeout)
  26. end_time = time.time()
  27. response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
  28. metrics["success"] = True
  29. metrics["response_time"] = end_time - start_time
  30. metrics["payload_size"] = len(response.content)
  31. try:
  32. payload = response.json()
  33. except json.JSONDecodeError:
  34. metrics["error_message"] = "Invalid JSON response"
  35. break
  36. return metrics # Exit on success
  37. except requests.exceptions.RequestException as e:
  38. metrics["error_message"] = str(e)
  39. if attempt < max_attempts - 1:
  40. print(f"Attempt {attempt + 1} failed: {e}. Retrying...")
  41. time.sleep(1) # Wait before retrying
  42. else:
  43. print(f"Max attempts reached. Failed to retrieve payload.")
  44. return metrics
  45. except Exception as e:
  46. metrics["error_message"] = str(e)
  47. return metrics # Return on unexpected error
  48. if __name__ == '__main__':
  49. # Example usage
  50. test_url = "https://jsonplaceholder.typicode.com/todos/1" # Replace with your URL
  51. metrics = collect_payload_metrics(test_url)
  52. print(metrics)
  53. test_url_invalid = "https://httpstat.us/500" #Example of a failure
  54. metrics = collect_payload_metrics(test_url_invalid)
  55. print(metrics)

Add your comment