File size: 6,992 Bytes
b7d2408 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
"""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")
|