Update app.py

app.py

@@ -6,7 +6,8 @@ import gradio as gr
 import requests
 import logging
 import time
-from typing import Optional
+from typing import Optional, Dict, Any
+import json
 
 # Configure logging
 logging.basicConfig(level=logging.DEBUG)
@@ -39,58 +40,99 @@ class TokenManager:
         self.last_obtained: float = 0
         self.token_expiry: int = 3600  # 1 hour default expiry
 
+    def _make_login_request(self, payload: Dict[str, Any]) -> Optional[str]:
+        """Helper method to make the login request with different approaches"""
+        approaches = [
+            self._try_json_request,
+            self._try_form_data_request,
+            self._try_urlencoded_request
+        ]
+        
+        for approach in approaches:
+            try:
+                result = approach(payload)
+                if result:
+                    return result
+            except Exception as e:
+                logger.warning(f"Approach failed: {approach.__name__}, error: {str(e)}")
+                continue
+                
+        return None
+
+    def _try_json_request(self, payload: Dict[str, Any]) -> Optional[str]:
+        """Try sending as JSON with proper headers"""
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json"
+        }
+        res = requests.post(
+            f"{BACKEND_URL}/auth/login",
+            json=payload,
+            headers=headers,
+            timeout=10
+        )
+        return self._process_response(res)
+
+    def _try_form_data_request(self, payload: Dict[str, Any]) -> Optional[str]:
+        """Try sending as form data"""
+        res = requests.post(
+            f"{BACKEND_URL}/auth/login",
+            data=payload,
+            timeout=10
+        )
+        return self._process_response(res)
+
+    def _try_urlencoded_request(self, payload: Dict[str, Any]) -> Optional[str]:
+        """Try sending as x-www-form-urlencoded"""
+        headers = {
+            "Content-Type": "application/x-www-form-urlencoded",
+        }
+        res = requests.post(
+            f"{BACKEND_URL}/auth/login",
+            data=payload,
+            headers=headers,
+            timeout=10
+        )
+        return self._process_response(res)
+
+    def _process_response(self, response: requests.Response) -> Optional[str]:
+        """Process the response and extract token if successful"""
+        if response.status_code == 200:
+            token_data = response.json()
+            return token_data.get("access_token")
+        logger.error(f"Request failed: {response.status_code} - {response.text}")
+        return None
+
     def get_token(self, force_refresh: bool = False) -> str:
         current_time = time.time()
         
         if not force_refresh and self.token and (current_time - self.last_obtained) < self.token_expiry:
             return self.token
             
+        login_payload = {
+            "username": ADMIN_EMAIL,
+            "password": ADMIN_PASSWORD,
+            "device_token": "admin-device-token"
+        }
+        
         for attempt in range(MAX_TOKEN_RETRIES):
             try:
-                login_payload = {
-                    "username": ADMIN_EMAIL,
-                    "password": ADMIN_PASSWORD,
-                    "device_token": "admin-device-token"
-                }
-                
                 logger.debug(f"Attempting to obtain admin token (attempt {attempt + 1})")
                 
-                # Ensure proper headers for JSON content
-                headers = {
-                    "Content-Type": "application/json",
-                    "Accept": "application/json"
-                }
-                
-                res = requests.post(
-                    f"{BACKEND_URL}/auth/login",
-                    json=login_payload,  # Using json parameter automatically sets content-type
-                    headers=headers,
-                    timeout=10
-                )
-                
-                logger.debug(f"Response status: {res.status_code}, content: {res.text}")
-                
-                if res.status_code == 200:
-                    token_data = res.json()
-                    self.token = token_data.get("access_token")
-                    if not self.token:
-                        raise Exception("No access_token in response")
-                    
+                token = self._make_login_request(login_payload)
+                if token:
+                    self.token = token
                     self.last_obtained = current_time
-                    self.token_expiry = token_data.get("expires_in", 3600)
                     logger.info("Successfully obtained admin token")
                     return self.token
                 
-                logger.error(f"Token request failed: {res.status_code} - {res.text}")
-                if res.status_code == 422:
-                    raise Exception(f"Validation error: {res.json().get('detail', 'Unknown validation error')}")
-                if res.status_code == 401:
-                    raise Exception("Invalid admin credentials")
+                if attempt < MAX_TOKEN_RETRIES - 1:
+                    time.sleep(TOKEN_RETRY_DELAY * (attempt + 1))
                     
             except requests.exceptions.RequestException as e:
                 logger.error(f"Network error during token fetch: {str(e)}")
                 if attempt < MAX_TOKEN_RETRIES - 1:
-                    time.sleep(TOKEN_RETRY_DELAY * (attempt + 1))  # Exponential backoff
+                    time.sleep(TOKEN_RETRY_DELAY * (attempt + 1))
                 continue
             except Exception as e:
                 logger.error(f"Unexpected error during token fetch: {str(e)}")
@@ -98,10 +140,11 @@ class TokenManager:
                     time.sleep(TOKEN_RETRY_DELAY)
                 continue
                 
-        raise Exception("Failed to obtain admin token after multiple attempts")
+        raise Exception("Failed to obtain admin token after multiple attempts and approaches")
 
 token_manager = TokenManager()
 
+
 @app.get("/")
 def root():
     logger.debug("Root endpoint accessed")