python-services/Retrieve/utils/http_util.py

"""
HTTP utility functions for making synchronous and asynchronous HTTP requests.

This module provides a unified interface for HTTP operations using httpx,
with proper error handling, timeout configuration, and retry logic.
"""

import asyncio
import os
import time
import uuid
from typing import Any, Dict, Optional

import httpx

from utils.bio_logger import bio_logger as logger


class HTTPError(Exception):
    """Custom exception for HTTP-related errors."""

    def __init__(self, status_code: int, message: str, url: str):
        self.status_code = status_code
        self.message = message
        self.url = url
        super().__init__(f"HTTP {status_code}: {message} for {url}")


def _create_timeout(timeout: float = 10.0) -> httpx.Timeout:
    """Create a timeout configuration for HTTP requests."""
    return httpx.Timeout(timeout, connect=5.0)


def _handle_response(response: httpx.Response, url: str) -> Any:
    """Handle HTTP response and raise appropriate exceptions."""
    if response.status_code == 200:
        return response.json()

    logger.error(f"HTTP request failed: {response.status_code} for {url}")
    raise HTTPError(
        status_code=response.status_code,
        message=f"Request failed with status {response.status_code}",
        url=url,
    )


async def async_http_get(
    url: str,
    params: Optional[Dict[str, Any]] = None,
    timeout: float = 10.0,
    headers: Optional[Dict[str, str]] = None,
) -> Any:
    """
    Make an asynchronous HTTP GET request.

    Args:
        url: The URL to make the request to
        params: Query parameters to include in the request
        timeout: Request timeout in seconds
        headers: Optional headers to include in the request

    Returns:
        The JSON response from the server

    Raises:
        HTTPError: If the request fails
        httpx.RequestError: If there's a network error
    """
    timeout_config = _create_timeout(timeout)
    start_time = time.time()

    async with httpx.AsyncClient(timeout=timeout_config) as client:
        response = await client.get(url=url, params=params, headers=headers)
        duration = time.time() - start_time

        # Log the API call
        logger.log_api_call("GET", url, response.status_code, duration)

        return _handle_response(response, url)


def http_get(
    url: str,
    params: Optional[Dict[str, Any]] = None,
    timeout: float = 10.0,
    headers: Optional[Dict[str, str]] = None,
) -> Any:
    """
    Make a synchronous HTTP GET request.

    Args:
        url: The URL to make the request to
        params: Query parameters to include in the request
        timeout: Request timeout in seconds
        headers: Optional headers to include in the request

    Returns:
        The JSON response from the server

    Raises:
        HTTPError: If the request fails
        httpx.RequestError: If there's a network error
    """
    timeout_config = _create_timeout(timeout)
    start_time = time.time()

    with httpx.Client(timeout=timeout_config) as client:
        response = client.get(url=url, params=params, headers=headers)
        duration = time.time() - start_time

        # Log the API call
        logger.log_api_call("GET", url, response.status_code, duration)

        return _handle_response(response, url)


def http_post(
    url: str, data: Any, headers: Optional[Dict[str, Any]] = None, timeout: float = 10.0
) -> Any:
    """
    Make a synchronous HTTP POST request.

    Args:
        url: The URL to make the request to
        data: The data to send in the request body
        headers: Optional headers to include in the request
        timeout: Request timeout in seconds

    Returns:
        The JSON response from the server

    Raises:
        HTTPError: If the request fails
        httpx.RequestError: If there's a network error
    """
    timeout_config = _create_timeout(timeout)
    start_time = time.time()

    with httpx.Client(timeout=timeout_config) as client:
        response = client.post(url=url, json=data, headers=headers)
        duration = time.time() - start_time

        # Log the API call
        logger.log_api_call("POST", url, response.status_code, duration)

        return _handle_response(response, url)


async def async_http_post(
    url: str,
    data: Any,
    headers: Optional[Dict[str, Any]] = None,
    timeout: float = 10.0,
    max_retries: int = 3,
    retry_delay: float = 0.5,
) -> Any:
    """
    Make an asynchronous HTTP POST request with retry logic.

    Args:
        url: The URL to make the request to
        data: The data to send in the request body
        headers: Optional headers to include in the request
        timeout: Request timeout in seconds
        max_retries: Maximum number of retry attempts
        retry_delay: Delay between retries in seconds

    Returns:
        The JSON response from the server

    Raises:
        HTTPError: If the request fails after all retries
        httpx.RequestError: If there's a network error
    """
    timeout_config = _create_timeout(timeout)

    async with httpx.AsyncClient(timeout=timeout_config) as client:
        for attempt in range(1, max_retries + 1):
            try:
                start_time = time.time()
                response = await client.post(url=url, json=data, headers=headers)
                duration = time.time() - start_time

                # Log the API call
                logger.log_api_call("POST", url, response.status_code, duration)

                if response.status_code == 200:
                    return response.json()
                else:
                    logger.error(
                        f"HTTP POST failed (attempt {attempt}/{max_retries}): "
                        f"{response.status_code} for {url}"
                    )
                    if attempt < max_retries:
                        await asyncio.sleep(retry_delay)
                    else:
                        raise HTTPError(
                            status_code=response.status_code,
                            message=f"Request failed after {max_retries} attempts",
                            url=url,
                        )
            except httpx.RequestError as e:
                logger.error(f"Network error on attempt {attempt}: {e}")
                if attempt < max_retries:
                    await asyncio.sleep(retry_delay)
                else:
                    raise HTTPError(
                        status_code=0,
                        message=f"Network error after {max_retries} attempts: {str(e)}",
                        url=url,
                    ) from e

    raise HTTPError(
        status_code=0,
        message=f"Failed to fetch data from {url} after {max_retries} attempts",
        url=url,
    )


def download_file(
    file_url: str, directory_path: str, timeout: int = 60, verify_ssl: bool = True
) -> Optional[str]:
    """
    Download a file from a URL to a local directory.

    Args:
        file_url: The URL of the file to download
        directory_path: The directory to save the file in
        timeout: Request timeout in seconds
        verify_ssl: Whether to verify SSL certificates

    Returns:
        The path to the downloaded file, or None if download failed
    """
    # Extract file extension from URL
    file_extension = file_url.split(".")[-1].split("?")[0]  # Remove query params
    random_filename = f"{uuid.uuid4()}.{file_extension}"

    # Create directory if it doesn't exist
    os.makedirs(directory_path, exist_ok=True)
    file_path = os.path.join(directory_path, random_filename)

    try:
        with httpx.Client(timeout=timeout, verify=verify_ssl) as client:
            with client.stream("GET", file_url) as response:
                if response.status_code == 200:
                    with open(file_path, "wb") as file:
                        for chunk in response.iter_bytes(chunk_size=8192):
                            file.write(chunk)
                    logger.info(f"Successfully downloaded file to {file_path}")
                    return file_path
                else:
                    logger.error(
                        f"Download failed with status code: {response.status_code}"
                    )
                    return None
    except httpx.TimeoutException:
        logger.error("Download request timed out")
        return None
    except httpx.RequestError as e:
        logger.error(f"Download request failed: {e}")
        return None
    except Exception as e:
        logger.error(f"Unexpected error during download: {e}")
        return None


# Backward compatibility functions
async def async_http_post_legacy(url: str, params: dict) -> Any:
    """
    Legacy async HTTP POST function for backward compatibility.

    This function maintains the old interface but uses the new implementation.
    """
    return await async_http_post(url=url, data=params)