163 lines
4.5 KiB
Python
163 lines
4.5 KiB
Python
"""Centralized HTTP client wrapper with retries and consistent error handling."""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Optional
|
|
|
|
import requests
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
|
|
from app.utils.logging_config import get_logger
|
|
|
|
|
|
@dataclass
|
|
class HttpResponse:
|
|
"""Wrapper for HTTP response data."""
|
|
|
|
status_code: int
|
|
json_data: Optional[dict[str, Any]]
|
|
text: str
|
|
success: bool
|
|
|
|
|
|
class HttpClient:
|
|
"""HTTP client with automatic retries, timeouts, and logging."""
|
|
|
|
def __init__(
|
|
self,
|
|
timeout: int = 30,
|
|
max_retries: int = 3,
|
|
backoff_factor: float = 0.5,
|
|
) -> None:
|
|
"""Initialize the HTTP client.
|
|
|
|
Args:
|
|
timeout: Request timeout in seconds.
|
|
max_retries: Maximum number of retry attempts.
|
|
backoff_factor: Multiplier for exponential backoff between retries.
|
|
"""
|
|
self.timeout = timeout
|
|
self.logger = get_logger(__name__)
|
|
|
|
# Configure retry strategy
|
|
retry_strategy = Retry(
|
|
total=max_retries,
|
|
backoff_factor=backoff_factor,
|
|
status_forcelist=[429, 500, 502, 503, 504],
|
|
allowed_methods=["GET", "POST"],
|
|
)
|
|
|
|
# Create session with retry adapter
|
|
self.session = requests.Session()
|
|
adapter = HTTPAdapter(max_retries=retry_strategy)
|
|
self.session.mount("http://", adapter)
|
|
self.session.mount("https://", adapter)
|
|
|
|
def get(
|
|
self,
|
|
url: str,
|
|
params: Optional[dict[str, Any]] = None,
|
|
headers: Optional[dict[str, str]] = None,
|
|
) -> HttpResponse:
|
|
"""Perform a GET request.
|
|
|
|
Args:
|
|
url: The URL to request.
|
|
params: Optional query parameters.
|
|
headers: Optional HTTP headers.
|
|
|
|
Returns:
|
|
HttpResponse with status, data, and success flag.
|
|
"""
|
|
self.logger.debug("http_get", url=url, params=params)
|
|
|
|
try:
|
|
response = self.session.get(
|
|
url,
|
|
params=params,
|
|
headers=headers,
|
|
timeout=self.timeout,
|
|
)
|
|
return self._build_response(response)
|
|
except requests.RequestException as e:
|
|
self.logger.error("http_get_failed", url=url, error=str(e))
|
|
return HttpResponse(
|
|
status_code=0,
|
|
json_data=None,
|
|
text=str(e),
|
|
success=False,
|
|
)
|
|
|
|
def post(
|
|
self,
|
|
url: str,
|
|
data: Optional[dict[str, Any]] = None,
|
|
json_data: Optional[dict[str, Any]] = None,
|
|
headers: Optional[dict[str, str]] = None,
|
|
) -> HttpResponse:
|
|
"""Perform a POST request.
|
|
|
|
Args:
|
|
url: The URL to request.
|
|
data: Optional form data.
|
|
json_data: Optional JSON data.
|
|
headers: Optional HTTP headers.
|
|
|
|
Returns:
|
|
HttpResponse with status, data, and success flag.
|
|
"""
|
|
self.logger.debug("http_post", url=url)
|
|
|
|
try:
|
|
response = self.session.post(
|
|
url,
|
|
data=data,
|
|
json=json_data,
|
|
headers=headers,
|
|
timeout=self.timeout,
|
|
)
|
|
return self._build_response(response)
|
|
except requests.RequestException as e:
|
|
self.logger.error("http_post_failed", url=url, error=str(e))
|
|
return HttpResponse(
|
|
status_code=0,
|
|
json_data=None,
|
|
text=str(e),
|
|
success=False,
|
|
)
|
|
|
|
def _build_response(self, response: requests.Response) -> HttpResponse:
|
|
"""Build an HttpResponse from a requests Response.
|
|
|
|
Args:
|
|
response: The requests library Response object.
|
|
|
|
Returns:
|
|
HttpResponse wrapper.
|
|
"""
|
|
json_data = None
|
|
if response.headers.get("content-type", "").startswith("application/json"):
|
|
try:
|
|
json_data = response.json()
|
|
except ValueError:
|
|
pass
|
|
|
|
success = 200 <= response.status_code < 300
|
|
|
|
self.logger.debug(
|
|
"http_response",
|
|
status_code=response.status_code,
|
|
success=success,
|
|
)
|
|
|
|
return HttpResponse(
|
|
status_code=response.status_code,
|
|
json_data=json_data,
|
|
text=response.text,
|
|
success=success,
|
|
)
|
|
|
|
def close(self) -> None:
|
|
"""Close the HTTP session."""
|
|
self.session.close()
|