chore: import upstream snapshot with attribution

shared/python/__init__.py

@@ -0,0 +1,29 @@
+"""
+Shared utilities for the Generative AI for Beginners course.
+
+This module provides common functionality used across multiple lessons,
+including environment variable handling, input validation, and API utilities.
+"""
+
+from .api_utils import (
+    create_azure_openai_client,
+    create_openai_client,
+    make_safe_request,
+)
+from .env_utils import get_required_env, validate_env_vars
+from .input_validation import (
+    sanitize_prompt_input,
+    validate_number_input,
+    validate_text_input,
+)
+
+__all__ = [
+    "get_required_env",
+    "validate_env_vars",
+    "validate_number_input",
+    "validate_text_input",
+    "sanitize_prompt_input",
+    "make_safe_request",
+    "create_openai_client",
+    "create_azure_openai_client",
+]

shared/python/api_utils.py

@@ -0,0 +1,174 @@
+"""
+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

shared/python/env_utils.py

@@ -0,0 +1,88 @@
+"""
+Environment variable utilities for secure configuration management.
+
+This module provides functions to safely retrieve and validate environment
+variables, ensuring that sensitive configuration is properly handled.
+"""
+
+import os
+
+
+def get_required_env(var_name: str, description: str | None = None) -> str:
+    """
+    Get a required environment variable or raise an error with helpful message.
+
+    Args:
+        var_name: The name of the environment variable to retrieve.
+        description: Optional description of what the variable is used for.
+
+    Returns:
+        The value of the environment variable.
+
+    Raises:
+        ValueError: If the environment variable is not set or is empty.
+
+    Example:
+        >>> api_key = get_required_env("OPENAI_API_KEY", "OpenAI API authentication")
+    """
+    value = os.getenv(var_name)
+    if not value:
+        desc_part = f" ({description})" if description else ""
+        raise ValueError(
+            f"Missing required environment variable: {var_name}{desc_part}. "
+            f"Please set it in your .env file or environment."
+        )
+    return value
+
+
+def validate_env_vars(*var_names: str) -> dict[str, str]:
+    """
+    Validate that multiple environment variables are set.
+
+    Args:
+        *var_names: Variable names to check.
+
+    Returns:
+        Dictionary mapping variable names to their values.
+
+    Raises:
+        ValueError: If any of the required variables are missing.
+
+    Example:
+        >>> env = validate_env_vars("AZURE_OPENAI_ENDPOINT", "AZURE_OPENAI_API_KEY")
+        >>> print(env["AZURE_OPENAI_ENDPOINT"])
+    """
+    missing = []
+    values = {}
+
+    for var_name in var_names:
+        value = os.getenv(var_name)
+        if not value:
+            missing.append(var_name)
+        else:
+            values[var_name] = value
+
+    if missing:
+        raise ValueError(
+            f"Missing required environment variables: {', '.join(missing)}. "
+            f"Please set them in your .env file or environment."
+        )
+
+    return values
+
+
+def get_env_with_default(var_name: str, default: str) -> str:
+    """
+    Get an environment variable with a default value.
+
+    Args:
+        var_name: The name of the environment variable.
+        default: The default value if the variable is not set.
+
+    Returns:
+        The value of the environment variable or the default.
+
+    Example:
+        >>> model = get_env_with_default("MODEL_NAME", "gpt-4o")
+    """
+    return os.getenv(var_name, default)

shared/python/input_validation.py

@@ -0,0 +1,204 @@
+"""
+Input validation utilities for secure user input handling.
+
+This module provides functions to validate and sanitize user input,
+protecting against prompt injection and other input-based attacks.
+"""
+
+import re
+
+
+def validate_number_input(
+    value: str, min_val: int = 1, max_val: int = 100, field_name: str = "number"
+) -> int:
+    """
+    Validate and convert string input to an integer within bounds.
+
+    Args:
+        value: The string value to validate.
+        min_val: Minimum allowed value (inclusive).
+        max_val: Maximum allowed value (inclusive).
+        field_name: Name of the field for error messages.
+
+    Returns:
+        The validated integer value.
+
+    Raises:
+        ValueError: If the value is not a valid integer or is out of bounds.
+
+    Example:
+        >>> num = validate_number_input("5", min_val=1, max_val=20)
+        >>> print(num)  # 5
+    """
+    try:
+        num = int(value.strip())
+        if num < min_val or num > max_val:
+            raise ValueError(f"{field_name} must be between {min_val} and {max_val}, got {num}")
+        return num
+    except (ValueError, AttributeError) as e:
+        if "must be between" in str(e):
+            raise
+        raise ValueError(
+            f"Please enter a valid {field_name} between {min_val} and {max_val}"
+        ) from e
+
+
+def validate_text_input(
+    value: str,
+    max_length: int = 500,
+    min_length: int = 1,
+    allow_empty: bool = False,
+    field_name: str = "input",
+) -> str:
+    """
+    Validate and sanitize text input.
+
+    Args:
+        value: The string value to validate.
+        max_length: Maximum allowed length.
+        min_length: Minimum required length.
+        allow_empty: Whether to allow empty strings.
+        field_name: Name of the field for error messages.
+
+    Returns:
+        The validated and trimmed string.
+
+    Raises:
+        ValueError: If the value fails validation.
+
+    Example:
+        >>> text = validate_text_input("Hello World", max_length=100)
+    """
+    if value is None:
+        if allow_empty:
+            return ""
+        raise ValueError(f"{field_name} cannot be None")
+
+    trimmed = value.strip()
+
+    if not trimmed:
+        if allow_empty:
+            return ""
+        raise ValueError(f"{field_name} cannot be empty")
+
+    if len(trimmed) > max_length:
+        raise ValueError(
+            f"{field_name} is too long. Maximum {max_length} characters allowed, "
+            f"got {len(trimmed)}"
+        )
+
+    if len(trimmed) < min_length:
+        raise ValueError(f"{field_name} is too short. Minimum {min_length} characters required")
+
+    return trimmed
+
+
+def sanitize_prompt_input(value: str, max_length: int = 1000, strict: bool = False) -> str:
+    """
+    Sanitize user input intended for use in LLM prompts.
+
+    This function removes potentially dangerous characters and patterns
+    that could be used for prompt injection attacks.
+
+    Args:
+        value: The string to sanitize.
+        max_length: Maximum allowed length after sanitization.
+        strict: If True, only allow alphanumeric, spaces, and basic punctuation.
+
+    Returns:
+        The sanitized string.
+
+    Raises:
+        ValueError: If the input is too long or contains only invalid characters.
+
+    Example:
+        >>> safe_input = sanitize_prompt_input("Hello, world!")
+    """
+    if not value:
+        return ""
+
+    # Trim whitespace
+    sanitized = value.strip()
+
+    # Remove null bytes and control characters (except newlines and tabs)
+    sanitized = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", "", sanitized)
+
+    # Remove potentially dangerous template/injection patterns
+    dangerous_patterns = [
+        r"\{\{.*?\}\}",  # Template injection
+        r"\${.*?}",  # Variable substitution
+        r"<script.*?>.*?</script>",  # Script tags
+        r"javascript:",  # JavaScript URLs
+    ]
+
+    for pattern in dangerous_patterns:
+        sanitized = re.sub(pattern, "", sanitized, flags=re.IGNORECASE | re.DOTALL)
+
+    if strict:
+        # In strict mode, only allow safe characters
+        sanitized = re.sub(r"[^\w\s,.\'\"-?!@#$%&*()+=:;]", "", sanitized, flags=re.UNICODE)
+
+    # Normalize whitespace
+    sanitized = re.sub(r"\s+", " ", sanitized)
+    sanitized = sanitized.strip()
+
+    if len(sanitized) > max_length:
+        raise ValueError(f"Input too long. Maximum {max_length} characters allowed.")
+
+    if not sanitized:
+        raise ValueError("Input contains only invalid characters")
+
+    return sanitized
+
+
+def validate_email(email: str) -> str:
+    """
+    Validate an email address format.
+
+    Args:
+        email: The email address to validate.
+
+    Returns:
+        The validated email address (lowercase).
+
+    Raises:
+        ValueError: If the email format is invalid.
+    """
+    email = email.strip().lower()
+
+    # Basic email pattern
+    pattern = r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$"
+
+    if not re.match(pattern, email):
+        raise ValueError(f"Invalid email format: {email}")
+
+    return email
+
+
+def validate_url(url: str, require_https: bool = True) -> str:
+    """
+    Validate a URL format.
+
+    Args:
+        url: The URL to validate.
+        require_https: If True, only allow HTTPS URLs.
+
+    Returns:
+        The validated URL.
+
+    Raises:
+        ValueError: If the URL format is invalid.
+    """
+    url = url.strip()
+
+    # Basic URL pattern
+    if require_https:
+        pattern = r"^https://[a-zA-Z0-9.-]+(?:/[^\s]*)?$"
+        if not re.match(pattern, url):
+            raise ValueError(f"Invalid HTTPS URL: {url}")
+    else:
+        pattern = r"^https?://[a-zA-Z0-9.-]+(?:/[^\s]*)?$"
+        if not re.match(pattern, url):
+            raise ValueError(f"Invalid URL: {url}")
+
+    return url