Source code for src.adapter

import json
import urllib.request
import time
from .response import Response




class BaseAdapter:
    """Abstract base class for HTTP adapters

    Defines the interface that all HTTP adapter implementations must follow
    Adapters are responsible for executing the actual HTTP request and returning a Response object
    """



    def request(self, method, url, **kwargs):
        """Execute an HTTP request

        Args:
            method (str): HTTP method (GET, POST, and so on)
            url (str): URL to request
            **kwargs: Additional arguments for the adapter

        Returns:
            Response: Response object containing status, headers, and body

        Raises:
            NotImplementedError: Must be implemented by subclass
        """
        raise NotImplementedError("Subclass must implement abstract method")






class UrllibAdapter(BaseAdapter):
    """HTTP adapter using Python's built-in urllib library.

    Provides a lightweight HTTP adapter without external dependencies.
    **Note**: Does not support connection pooling
    """



    def request(
        self,
        method: str,
        url: str,
        headers: dict = None,
        data: dict = None,
        timeout: int = 10,
    ) -> Response:
        """Execute an HTTP request using urllib.

        Args:
            method (str): HTTP method (GET, POST, PUT, DELETE, etc.)
            url (str): URL to request
            headers (dict, optional): HTTP headers. Defaults to None
            data (dict, optional): Request body data. Defaults to None
            timeout (int, optional): Request timeout in seconds. Defaults to 10

        Returns:
            Response: Response object with status code, headers, body, and elapsed time
        """
        # urllib does not support proper pooling and it will be leave it as is
        # if we tend to support pooling, it's not worth to use urllib
        headers = headers or {}
        start_time = time.perf_counter()

        if isinstance(data, dict):
            data = json.dumps(data).encode("utf-8")
            headers["Content-Type"] = "application/json"

        request = urllib.request.Request(url=url, data=data, headers=headers, method=method)

        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:
                elapsed_time = time.perf_counter() - start_time
                return Response(
                    status_code=response.status,
                    headers=dict(response.headers),
                    body=response.read(),
                    elapsed=elapsed_time,
                )
        except urllib.error.HTTPError as e:
            return Response(status_code=e.code, headers=dict(e.headers), body=e.read())






class RequestsAdapter(BaseAdapter):
    """HTTP adapter using the requests library

    Provides connection pooling and advanced HTTP features through the requests library
    """

    def __init__(self, pool_connections: int = 10, pool_maxsize: int = 50) -> None:
        """Initialize RequestsAdapter with connection pooling configuration.

        Args:
            pool_connections (int, optional): Number of connection pools. Defaults to 10.
            pool_maxsize (int, optional): Maximum size of connection pool. Defaults to 50.

        Raises:
            ImportError: If requests library is not installed
        """
        try:
            import requests
            from requests.adapters import HTTPAdapter
        except ImportError:
            raise ImportError(
                "It seems you haven't installed requests library. Please install it using `pip install requests`"
            )

        self.requests = requests
        self.session = requests.Session()

        # mount default requests adapter
        adapter = HTTPAdapter(pool_connections=pool_connections, pool_maxsize=pool_maxsize)

        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)



    def request(
        self,
        method: str,
        url: str,
        headers: dict = None,
        data: dict = None,
        timeout: int = None,
    ) -> Response:
        """Execute an HTTP request using requests library.

        Args:
            method (str): HTTP method (GET, POST, PUT, DELETE, etc.)
            url (str): URL to request
            headers (dict, optional): HTTP headers. Defaults to None.
            data (dict, optional): Request body data. Defaults to None.
            timeout (int, optional): Request timeout in seconds. Defaults to None.

        Returns:
            Response: Response object with status code, headers, body, and elapsed time
        """
        start_time = time.perf_counter()
        try:
            response = self.requests.request(
                method=method,
                url=url,
                headers=headers,
                json=data if isinstance(data, dict) else None,
                data=data if not isinstance(data, dict) else None,
                timeout=timeout,
            )
            elapsed_time = time.perf_counter() - start_time

            return Response(
                status_code=response.status_code,
                headers=response.headers,
                body=response.content,
                elapsed=elapsed_time,
            )
        except Exception as e:
            return Response(status_code=0, headers={}, body=str(e).encode("utf-8"))




    def close(self) -> None:
        """Close the session and clean up connection pools.

        Closes the underlying requests session, releasing all pooled connections
        and resources.
        """
        # clean up session such as close connection pool or adapters
        self.session.close()