File size: 4,799 Bytes
b30f068 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | """
Base API Client
Abstract base class for all API clients
"""
from abc import ABC, abstractmethod
import requests
from typing import Dict, Any, Optional
import time
class BaseAPIClient(ABC):
"""
Abstract base class for API clients
Provides common functionality:
- HTTP request handling
- Error handling
- Rate limiting
- Response validation
"""
def __init__(self):
"""Initialize base client"""
self.session = requests.Session()
self.timeout = 30 # Default timeout in seconds
self.last_request_time = 0
self.min_request_interval = 0.1 # 100ms between requests
def _wait_for_rate_limit(self):
"""Implement simple rate limiting"""
elapsed = time.time() - self.last_request_time
if elapsed < self.min_request_interval:
time.sleep(self.min_request_interval - elapsed)
self.last_request_time = time.time()
def _make_request(
self,
method: str,
url: str,
params: Optional[Dict] = None,
headers: Optional[Dict] = None,
data: Optional[Dict] = None,
json: Optional[Dict] = None
) -> requests.Response:
"""
Make HTTP request with error handling
Args:
method: HTTP method (GET, POST, etc.)
url: Request URL
params: Query parameters
headers: Request headers
data: Form data
json: JSON data
Returns:
Response object
Raises:
requests.exceptions.RequestException: On request failure
"""
self._wait_for_rate_limit()
try:
response = self.session.request(
method=method,
url=url,
params=params,
headers=headers,
data=data,
json=json,
timeout=self.timeout
)
response.raise_for_status()
return response
except requests.exceptions.Timeout:
raise requests.exceptions.RequestException(
f"Request to {url} timed out after {self.timeout}s"
)
except requests.exceptions.ConnectionError:
raise requests.exceptions.RequestException(
f"Failed to connect to {url}. Check your internet connection."
)
except requests.exceptions.HTTPError as e:
status_code = e.response.status_code
if status_code == 401:
raise requests.exceptions.RequestException(
"Authentication failed. Check your API key."
)
elif status_code == 429:
raise requests.exceptions.RequestException(
"Rate limit exceeded. Please wait before making more requests."
)
elif status_code == 404:
raise requests.exceptions.RequestException(
"Resource not found. Check your request parameters."
)
else:
raise requests.exceptions.RequestException(
f"HTTP {status_code} error: {str(e)}"
)
def get(self, url: str, params: Optional[Dict] = None,
headers: Optional[Dict] = None) -> Dict[str, Any]:
"""
Make GET request
Args:
url: Request URL
params: Query parameters
headers: Request headers
Returns:
JSON response as dict
"""
response = self._make_request("GET", url, params=params, headers=headers)
return response.json()
def post(self, url: str, data: Optional[Dict] = None,
json: Optional[Dict] = None, headers: Optional[Dict] = None) -> Dict[str, Any]:
"""
Make POST request
Args:
url: Request URL
data: Form data
json: JSON data
headers: Request headers
Returns:
JSON response as dict
"""
response = self._make_request("POST", url, data=data, json=json, headers=headers)
return response.json()
@abstractmethod
def _validate_params(self, **kwargs) -> bool:
"""
Validate request parameters
Should be implemented by subclasses
Args:
**kwargs: Parameters to validate
Returns:
True if valid
Raises:
ValueError: If parameters are invalid
"""
pass
def __del__(self):
"""Cleanup: close session"""
if hasattr(self, 'session'):
self.session.close()
|