gh-christianlouis-docuelevate/app/api/openai.py

"""
AI provider and OpenAI API endpoints.

Exposes two endpoints:
- GET /api/ai/test   – tests the currently configured AI provider (generic, provider-agnostic)
- GET /api/openai/test – backward-compatible alias that tests the OpenAI API specifically
"""

import logging

from fastapi import APIRouter, Request

from app.auth import require_login
from app.config import settings

# Set up logging
logger = logging.getLogger(__name__)

router = APIRouter()


def _get_exception_chain_detail(exc: Exception) -> str:
    """
    Extract a verbose diagnostic message by walking the full exception chain.

    Surfaces DNS resolution failures, TCP connection refused errors, SSL issues,
    and other low-level network problems that are normally hidden behind a generic
    'Connection error.' message.
    """
    parts: list[str] = [str(exc)]

    cause = getattr(exc, "__cause__", None) or getattr(exc, "__context__", None)
    seen: set[int] = {id(exc)}
    while cause is not None and id(cause) not in seen:
        seen.add(id(cause))
        cause_str = str(cause)
        if cause_str and cause_str not in parts:
            parts.append(f"caused by: {type(cause).__name__}: {cause_str}")
        cause = getattr(cause, "__cause__", None) or getattr(cause, "__context__", None)

    return " | ".join(parts)


@router.get("/openai/test")
@require_login
async def test_openai_connection(request: Request):
    """
    Test if the configured OpenAI API key is valid.
    """
    try:
        import openai

        logger.info("Testing OpenAI API key validity")

        # Check if API key is configured
        if not settings.openai_api_key:
            logger.warning("No OpenAI API key configured")
            return {"status": "error", "message": "No OpenAI API key is configured"}

        # Configure the client, explicitly passing base_url so the sanitized
        # value from Settings (strip_outer_quotes) is used instead of the raw
        # OPENAI_BASE_URL env var which may contain literal quote characters.
        client = openai.OpenAI(api_key=settings.openai_api_key, base_url=settings.openai_base_url)

        # Try to make a simple request to validate the key
        try:
            # Use a models list endpoint as a simple validation
            models = client.models.list()

            # If we got here, the key is valid
            logger.info("OpenAI API key is valid")
            return {
                "status": "success",
                "message": "OpenAI API key is valid",
                "models_available": len(models.data) if hasattr(models, "data") else "Unknown",
            }

        except openai.APITimeoutError as e:
            detail = _get_exception_chain_detail(e)
            logger.error(f"OpenAI API request timed out: {detail}", exc_info=True)
            return {
                "status": "error",
                "message": f"Request timed out: {detail}",
                "is_auth_error": False,
                "error_type": "timeout",
            }

        except openai.APIConnectionError as e:
            detail = _get_exception_chain_detail(e)
            base_url = getattr(getattr(client, "_client", None), "base_url", None)
            base_url_info = f" (base_url: {base_url})" if base_url else ""
            logger.error(
                f"OpenAI API connection error{base_url_info}: {detail}",
                exc_info=True,
            )
            return {
                "status": "error",
                "message": f"Connection error{base_url_info}: {detail}",
                "is_auth_error": False,
                "error_type": "connection_error",
            }

        except openai.AuthenticationError as e:
            logger.error(f"OpenAI authentication error (status {e.status_code}): {e.message}", exc_info=True)
            return {
                "status": "error",
                "message": f"Authentication failed: {e.message}",
                "is_auth_error": True,
                "error_type": "authentication_error",
            }

        except openai.RateLimitError as e:
            logger.warning(f"OpenAI rate limit exceeded (status {e.status_code}): {e.message}")
            return {
                "status": "error",
                "message": f"Rate limit exceeded: {e.message}",
                "is_auth_error": False,
                "error_type": "rate_limit",
            }

        except openai.APIStatusError as e:
            logger.error(
                f"OpenAI API returned HTTP {e.status_code}: {e.message} | "
                f"request_id={e.response.headers.get('x-request-id', 'n/a')}"
            )
            return {
                "status": "error",
                "message": f"API error (HTTP {e.status_code}): {e.message}",
                "is_auth_error": e.status_code == 401,
                "error_type": "api_status_error",
                "http_status": e.status_code,
            }

        except Exception as e:
            error_msg = str(e)
            logger.error(f"OpenAI API key test failed: {error_msg}", exc_info=True)

            # Determine if this is an authentication error
            is_auth_error = "auth" in error_msg.lower() or "api key" in error_msg.lower()

            return {
                "status": "error",
                "message": f"API key validation failed: {error_msg}",
                "is_auth_error": is_auth_error,
            }

    except ImportError:
        logger.exception("OpenAI package not installed")
        return {"status": "error", "message": "OpenAI package not installed"}
    except Exception as e:
        logger.exception("Unexpected error testing OpenAI connection")
        return {"status": "error", "message": f"Unexpected error: {str(e)}"}


@router.get("/ai/test")
@require_login
async def test_ai_provider_connection(request: Request):
    """
    Test the currently configured AI provider connection.

    Uses ``get_ai_provider()`` to instantiate the active provider and sends a
    minimal chat completion to verify that the credentials and endpoint are
    reachable.  Works for all supported providers (OpenAI, Azure, Anthropic,
    Gemini, Ollama, OpenRouter, Portkey, LiteLLM).
    """
    from app.utils.ai_provider import get_ai_provider

    provider_name = settings.ai_provider
    model = settings.ai_model or settings.openai_model

    logger.info(f"Testing AI provider connection: provider={provider_name}, model={model}")

    try:
        provider = get_ai_provider()
        response = provider.chat_completion(
            messages=[{"role": "user", "content": "Reply with the single word: ok"}],
            model=model,
            temperature=0,
            max_tokens=5,
        )
        logger.info(f"AI provider test successful: provider={provider_name}")
        return {
            "status": "success",
            "message": f"AI provider '{provider_name}' is reachable and responding",
            "provider": provider_name,
            "model": model,
            "response_preview": (response or "")[:50],
        }
    except ValueError as e:
        # Configuration errors (missing keys, unknown provider)
        logger.warning(f"AI provider configuration error: {e}")
        return {
            "status": "error",
            "message": str(e),
            "provider": provider_name,
        }
    except Exception as e:
        detail = _get_exception_chain_detail(e)
        logger.error(f"AI provider test failed for '{provider_name}': {detail}", exc_info=True)
        return {
            "status": "error",
            "message": f"Connection failed: {detail}",
            "provider": provider_name,
        }