metrics_evaluation/cvat_api/auth.py

"""CVAT API authentication methods."""

import time
from typing import TYPE_CHECKING

import requests

# TYPE_CHECKING is False at runtime but True during static type checking.
# This allows us to import CvatApiClient for type hints without creating a circular
# import (client.py imports AuthMethods, and we need CvatApiClient for type hints).
# Benefits:
# - Avoids circular import errors at runtime
# - Provides proper type checking during development
# - No performance overhead (import only happens during type checking, not at runtime)
# This is a recommended pattern in modern Python (PEP 484, PEP 563).
# -- Claude Code
if TYPE_CHECKING:
    from .client import CvatApiClient


class AuthMethods:
    """Authentication methods for CVAT API."""

    def __init__(self, client: "CvatApiClient"):
        """Initialize auth methods with client reference.

        Args:
            client: Parent CvatApiClient instance
        """
        self.client = client

    def get_auth_token(self) -> str:
        """Authenticate with the CVAT server and return the token.

        Automatically retries on transient errors with exponential backoff:
        - Network timeouts
        - Connection errors
        - HTTP 502 (Bad Gateway), 503 (Service Unavailable), 504 (Gateway Timeout)

        Returns:
            Authentication token string

        Raises:
            TimeoutError: If authentication request times out after all retries
            ConnectionError: If authentication fails with HTTP error after all retries
            RuntimeError: If authentication fails due to network error after all retries
            ValueError: If no token received or invalid response
        """
        auth_url = f"{self.client.cvat_host}/api/auth/login"
        json_data = {
            "username": self.client.cvat_username,
            "password": self.client.cvat_password,
        }

        # Transient error status codes that should trigger retry
        TRANSIENT_STATUS_CODES = {502, 503, 504}

        last_exception = None

        for attempt in range(self.client.max_retries + 1):
            if attempt > 0:
                # Calculate exponential backoff delay
                delay = min(
                    self.client.initial_retry_delay * (2 ** (attempt - 1)),
                    self.client.max_retry_delay
                )
                self.client.logger.warning(
                    "⏳ Retry %d/%d for authentication after %.1fs delay...",
                    attempt,
                    self.client.max_retries,
                    delay
                )
                time.sleep(delay)

            # Routine log - no emoji (only for milestones)
            self.client.logger.debug(
                "Attempting authentication with CVAT server at %s (attempt %d)",
                self.client.cvat_host,
                attempt + 1
            )

            try:
                response = requests.post(
                    auth_url, json=json_data, timeout=self.client.cvat_auth_timeout
                )
                response.raise_for_status()

                token = response.json().get("key")
                if not token:
                    self.client.logger.error(
                        "❌ Authentication failed: No token received from the server"
                    )
                    raise ValueError(
                        "❌ Authentication failed: No token received from the server."
                    )

                self.client.logger.info("✅ Successfully authenticated with CVAT server")
                return token

            except requests.Timeout as e:
                last_exception = e
                self.client.logger.warning(
                    "⚠️  Authentication timeout (attempt %d/%d)",
                    attempt + 1,
                    self.client.max_retries + 1
                )
                # Continue to retry

            except requests.ConnectionError as e:
                last_exception = e
                self.client.logger.warning(
                    "⚠️  Authentication connection error (attempt %d/%d)",
                    attempt + 1,
                    self.client.max_retries + 1
                )
                # Continue to retry

            except requests.HTTPError as e:
                last_exception = e
                # Only retry on transient HTTP errors
                if hasattr(e, "response") and e.response.status_code in TRANSIENT_STATUS_CODES:
                    self.client.logger.warning(
                        "⚠️  Transient HTTP %d error during authentication (attempt %d/%d)",
                        e.response.status_code,
                        attempt + 1,
                        self.client.max_retries + 1
                    )
                    # Continue to retry
                else:
                    # Non-transient error, don't retry
                    self.client.logger.error(
                        "❌ Authentication failed: %d", e.response.status_code
                    )
                    self.client._handle_response_errors(
                        e.response, "❌ Authentication failed"
                    )
                    raise ConnectionError(
                        f"❌ Authentication failed: {e.response.status_code} - {e.response.text}"
                    ) from e

            except requests.RequestException as e:
                # Other request exceptions - don't retry
                last_exception = e
                self.client.logger.error(
                    "❌ Authentication request failed due to a network error: %s", e
                )
                raise RuntimeError(
                    "❌ Authentication request failed due to a network error."
                ) from e

            except ValueError as e:
                # No token in response - don't retry
                raise

            except Exception as e:
                # Other exceptions - don't retry
                self.client.logger.error("❌ Authentication failed: %s", e)
                raise ValueError(
                    "❌ Authentication failed: Invalid response from server."
                ) from e

        # All retries exhausted
        self.client.logger.error(
            "❌ All %d authentication attempts exhausted",
            self.client.max_retries + 1
        )
        if last_exception:
            if isinstance(last_exception, requests.Timeout):
                raise TimeoutError("❌ Authentication request timed out after all retries.") from last_exception
            elif isinstance(last_exception, requests.ConnectionError):
                raise RuntimeError("❌ Authentication connection failed after all retries.") from last_exception
            else:
                raise last_exception

        # Should never reach here, but just in case
        raise RuntimeError(f"Authentication failed after {self.client.max_retries + 1} attempts")