1. import requests
  2. import time
  3. import logging
  4. from typing import Dict, Any
  6. # Configure logging
  7. logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
  9. class APIAnomalyDetector:
  10. def __init__(self, api_endpoint: str, expected_status_code: int, retry_count: int = 3, retry_delay: int = 5):
  11. """
  12. Initializes the API Anomaly Detector.
  14. Args:
  15. api_endpoint: The URL of the API endpoint to monitor.
  16. expected_status_code: The expected HTTP status code for a healthy endpoint.
  17. retry_count: The number of times to retry the request if it fails.
  18. retry_delay: The delay (in seconds) between retries.
  19. """
  20. self.api_endpoint = api_endpoint
  21. self.expected_status_code = expected_status_code
  22. self.retry_count = retry_count
  23. self.retry_delay = retry_delay
  25. def check_endpoint(self) -> bool:
  26. """
  27. Checks the API endpoint and flags anomalies based on status code.
  29. Returns:
  30. True if the endpoint is healthy (status code matches), False otherwise (anomaly detected).
  31. """
  32. for attempt in range(self.retry_count):
  33. try:
  34. response = requests.get(self.api_endpoint)
  35. response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
  37. if response.status_code == self.expected_status_code:
  38. logging.info(f"Endpoint {self.api_endpoint} is healthy (Status Code: {response.status_code})")
  39. return True
  40. else:
  41. logging.warning(f"Endpoint {self.api_endpoint} returned unexpected status code: {response.status_code}")
  42. return False #Anomaly Detected
  43. except requests.exceptions.RequestException as e:
  44. logging.error(f"Error accessing endpoint {self.api_endpoint}: {e}")
  45. if attempt < self.retry_count - 1:
  46. logging.info(f"Retrying in {self.retry_delay} seconds...")
  47. time.sleep(self.retry_delay)
  48. else:
  49. logging.critical(f"Endpoint {self.api_endpoint} is down after {self.retry_count} attempts.")
  50. return False #Anomaly Detected
  51. return False #Should not reach here, but added for safety
  53. if __name__ == '__main__':
  54. # Example Usage
  55. api_detector = APIAnomalyDetector(api_endpoint="https://api.example.com/data", expected_status_code=200)
  56. is_healthy = api_detector.check_endpoint()
  58. if not is_healthy:
  59. print("API Endpoint Anomaly Detected. Maintenance Required.")
  60. else:
  61. print("API Endpoint is Operational.")

Add your comment