microsoft--generative-ai-for-beginners/shared/python/api_utils.py

"""
API utilities for safe HTTP requests and OpenAI client creation.

This module provides wrapper functions for making HTTP requests with
proper timeout, error handling, and retry logic.
"""

import os
from typing import Any

import requests
from requests.exceptions import RequestException


def make_safe_request(
    url: str, method: str = "GET", timeout: int = 30, retries: int = 3, **kwargs: Any
) -> requests.Response:
    """
    Make an HTTP request with proper timeout and error handling.

    Args:
        url: The URL to request.
        method: HTTP method (GET, POST, etc.).
        timeout: Request timeout in seconds.
        retries: Number of retry attempts.
        **kwargs: Additional arguments to pass to requests.

    Returns:
        The Response object.

    Raises:
        RequestException: If the request fails after all retries.

    Example:
        >>> response = make_safe_request("https://api.example.com/data")
        >>> data = response.json()
    """
    last_exception: Exception | None = None

    for attempt in range(retries):
        try:
            response = requests.request(method=method, url=url, timeout=timeout, **kwargs)
            response.raise_for_status()
            return response
        except RequestException as e:
            last_exception = e
            if attempt < retries - 1:
                # Exponential backoff could be added here
                continue
            raise

    # This should never be reached, but just in case
    raise last_exception or RequestException("Request failed")


def create_openai_client(api_key: str | None = None) -> Any:
    """
    Create an OpenAI client with proper configuration.

    Args:
        api_key: Optional API key. If not provided, reads from OPENAI_API_KEY env var.

    Returns:
        An OpenAI client instance.

    Raises:
        ValueError: If no API key is available.
        ImportError: If the openai package is not installed.

    Example:
        >>> client = create_openai_client()
        >>> response = client.responses.create(model="gpt-4o-mini", input="Hello")
    """
    try:
        from openai import OpenAI
    except ImportError as e:
        raise ImportError(
            "The 'openai' package is required. Install it with: pip install openai"
        ) from e

    key = api_key or os.getenv("OPENAI_API_KEY")
    if not key:
        raise ValueError(
            "OpenAI API key is required. Set OPENAI_API_KEY environment variable "
            "or pass api_key parameter."
        )

    return OpenAI(api_key=key)


def create_azure_openai_client(
    endpoint: str | None = None,
    api_key: str | None = None,
) -> Any:
    """
    Create an Azure OpenAI (Microsoft Foundry) client with proper configuration.

    The client targets the Azure OpenAI v1 endpoint (``<endpoint>/openai/v1/``),
    which powers the Responses API. Because the v1 endpoint is used, no
    ``api_version`` is required.

    Args:
        endpoint: Azure OpenAI endpoint URL. If not provided, reads from
                  AZURE_OPENAI_ENDPOINT env var.
        api_key: Azure OpenAI API key. If not provided, reads from
                 AZURE_OPENAI_API_KEY env var.

    Returns:
        An OpenAI client instance configured for the Azure v1 endpoint.

    Raises:
        ValueError: If endpoint or API key is missing.
        ImportError: If the openai package is not installed.

    Example:
        >>> client = create_azure_openai_client()
        >>> response = client.responses.create(model="gpt-4o-mini", input="Hello")
    """
    try:
        from openai import OpenAI
    except ImportError as e:
        raise ImportError(
            "The 'openai' package is required. Install it with: pip install openai"
        ) from e

    _endpoint = endpoint or os.getenv("AZURE_OPENAI_ENDPOINT")
    _api_key = api_key or os.getenv("AZURE_OPENAI_API_KEY")

    if not _endpoint:
        raise ValueError(
            "Azure OpenAI endpoint is required. Set AZURE_OPENAI_ENDPOINT "
            "environment variable or pass endpoint parameter."
        )

    if not _api_key:
        raise ValueError(
            "Azure OpenAI API key is required. Set AZURE_OPENAI_API_KEY "
            "environment variable or pass api_key parameter."
        )

    return OpenAI(
        api_key=_api_key,
        base_url=f"{_endpoint.rstrip('/')}/openai/v1/",
    )


def download_image(url: str, save_path: str, timeout: int = 30) -> str:
    """
    Download an image from a URL and save it to disk.

    Args:
        url: The URL of the image to download.
        save_path: The path where the image should be saved.
        timeout: Request timeout in seconds.

    Returns:
        The path where the image was saved.

    Raises:
        RequestException: If the download fails.
        IOError: If the file cannot be written.

    Example:
        >>> path = download_image("https://example.com/image.png", "./image.png")
    """
    response = make_safe_request(url, timeout=timeout)

    # Ensure the directory exists
    os.makedirs(os.path.dirname(save_path) or ".", exist_ok=True)

    with open(save_path, "wb") as f:
        f.write(response.content)

    return save_path