A reusable HTTP client wrapper for making API requests with default settings.

HTTPClient.py

"""
http_client.py

A reusable HTTP client wrapper for making API requests with default settings.

Features:
- Supports GET, POST, PUT, and DELETE requests.
- Automatically applies a base URL, timeout, and default headers.
- Handles request errors gracefully.

Usage:
    client = HTTPClient(base_url="https://api.ryadpasha.com", timeout=15, headers={"Authorization": "Bearer token"})
    response = client.get("/account/12345")
    print(response)

Author: Mohamed Riyad
Date: 13 Mar, 2025
"""

import requests
from aws_lambda_powertools import Tracer

tracer = Tracer()

class HTTPClient:
    def __init__(self, base_url, timeout=30, headers=None):
        """
        Initializes the HTTP client with default configurations.

        :param base_url: Base URL for all API requests.
        :param timeout: Default timeout for requests (in seconds).
        :param headers: Default headers for all requests.
        """
        self.base_url = base_url.rstrip("/")  # Ensure no trailing slash
        self.timeout = timeout
        self.headers = headers or {}

    @tracer.capture_method
    def _request(self, method, endpoint, **kwargs):
        """
        Generic method to handle HTTP requests.

        :param method: HTTP method (GET, POST, PUT, DELETE, etc.).
        :param endpoint: API endpoint (appended to base_url).
        :param kwargs: Additional arguments like params, json, data, headers.
        :return: Standardized JSON response.
        """
        url = f"{self.base_url}/{endpoint.lstrip('/')}"  # Ensure correct URL formatting
        headers = {**self.headers, **(kwargs.pop("headers", {}) or {})}  # Merge default & custom headers

        response = None

        try:
            response = requests.request(method, url, timeout=self.timeout, headers=headers, **kwargs)
            response.raise_for_status()  # Raise an error for bad responses (4xx, 5xx)

            return {
                "success": True,
                "data": response.json()
            }

        except requests.exceptions.HTTPError as e:
            return {
                "success": False,
                "errorCode": self._map_error_code(response.status_code),
                "errorMessage": response.text if response else str(e)
            }
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "errorCode": "InternalError",
                "errorMessage": response.text if response else str(e)
            }

    def _map_error_code(self, status_code):
        """Maps HTTP status codes to specific error codes."""
        if 400 <= status_code < 500:
            return "InvalidRequest"
        elif status_code >= 500:
            return "InternalError"
        return "ValidationError"

    def get(self, endpoint, params=None, headers=None):
        """Handles GET requests."""
        return self._request("GET", endpoint, params=params, headers=headers)

    def post(self, endpoint, json=None, data=None, headers=None):
        """Handles POST requests."""
        return self._request("POST", endpoint, json=json, data=data, headers=headers)

    def put(self, endpoint, json=None, data=None, headers=None):
        """Handles PUT requests."""
        return self._request("PUT", endpoint, json=json, data=data, headers=headers)

    def delete(self, endpoint, headers=None):
        """Handles DELETE requests."""
        return self._request("DELETE", endpoint, headers=headers)