"""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")