Ali2206 commited on
Commit
61dcdad
Β·
verified Β·
1 Parent(s): f2d2b0c

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +26 -17
app.py CHANGED
@@ -6,6 +6,7 @@ import gradio as gr
6
  import requests
7
  import logging
8
  import time
 
9
 
10
  # Configure logging
11
  logging.basicConfig(level=logging.DEBUG)
@@ -32,14 +33,13 @@ ADMIN_PASSWORD = "123456"
32
  MAX_TOKEN_RETRIES = 3
33
  TOKEN_RETRY_DELAY = 2 # seconds
34
 
35
- # Global token storage with refresh capability
36
  class TokenManager:
37
  def __init__(self):
38
- self.token = None
39
- self.last_obtained = 0
40
- self.token_expiry = 3600 # 1 hour default expiry
41
 
42
- def get_token(self, force_refresh=False):
43
  current_time = time.time()
44
 
45
  if not force_refresh and self.token and (current_time - self.last_obtained) < self.token_expiry:
@@ -54,12 +54,22 @@ class TokenManager:
54
  }
55
 
56
  logger.debug(f"Attempting to obtain admin token (attempt {attempt + 1})")
 
 
 
 
 
 
 
57
  res = requests.post(
58
  f"{BACKEND_URL}/auth/login",
59
- json=login_payload,
 
60
  timeout=10
61
  )
62
 
 
 
63
  if res.status_code == 200:
64
  token_data = res.json()
65
  self.token = token_data.get("access_token")
@@ -67,17 +77,23 @@ class TokenManager:
67
  raise Exception("No access_token in response")
68
 
69
  self.last_obtained = current_time
70
- # Get expiry from token if available, otherwise use default
71
  self.token_expiry = token_data.get("expires_in", 3600)
72
  logger.info("Successfully obtained admin token")
73
  return self.token
74
 
75
  logger.error(f"Token request failed: {res.status_code} - {res.text}")
 
 
76
  if res.status_code == 401:
77
  raise Exception("Invalid admin credentials")
78
 
79
  except requests.exceptions.RequestException as e:
80
  logger.error(f"Network error during token fetch: {str(e)}")
 
 
 
 
 
81
  if attempt < MAX_TOKEN_RETRIES - 1:
82
  time.sleep(TOKEN_RETRY_DELAY)
83
  continue
@@ -86,29 +102,24 @@ class TokenManager:
86
 
87
  token_manager = TokenManager()
88
 
89
- # Root endpoint
90
  @app.get("/")
91
  def root():
92
  logger.debug("Root endpoint accessed")
93
  return {"message": "πŸš€ FastAPI with MongoDB + JWT is running."}
94
 
95
- # Redirect /login to /auth/login
96
  @app.post("/login")
97
  async def redirect_login(request: Request):
98
  logger.info("Redirecting /login to /auth/login")
99
  return RedirectResponse(url="/auth/login", status_code=307)
100
 
101
- # Admin authentication logic
102
  def authenticate_admin(email: str = None, password: str = None):
103
  if email != ADMIN_EMAIL or password != ADMIN_PASSWORD:
104
  logger.warning(f"Failed admin login attempt with email: {email}")
105
  raise HTTPException(status_code=401, detail="Unauthorized: Invalid email or password")
106
-
107
  logger.info(f"Admin authenticated successfully: {email}")
108
  return True
109
 
110
- # Doctor creation with token handling
111
- def create_doctor(full_name, email, matricule, password, specialty):
112
  try:
113
  token = token_manager.get_token()
114
 
@@ -119,6 +130,7 @@ def create_doctor(full_name, email, matricule, password, specialty):
119
  "password": password,
120
  "specialty": specialty,
121
  }
 
122
  headers = {
123
  "Authorization": f"Bearer {token}",
124
  "Content-Type": "application/json"
@@ -146,7 +158,7 @@ def create_doctor(full_name, email, matricule, password, specialty):
146
  if res.status_code == 201:
147
  return "βœ… Doctor created successfully!"
148
 
149
- error_detail = res.json().get('detail', 'Unknown error occurred')
150
  return f"❌ Error: {error_detail} (Status: {res.status_code})"
151
 
152
  except requests.exceptions.RequestException as e:
@@ -154,7 +166,6 @@ def create_doctor(full_name, email, matricule, password, specialty):
154
  except Exception as e:
155
  return f"❌ System Error: {str(e)}"
156
 
157
- # Gradio interface
158
  admin_ui = gr.Blocks(css="""
159
  .gradio-container {
160
  background-color: #1A1B1F;
@@ -195,10 +206,8 @@ with admin_ui:
195
  outputs=output
196
  )
197
 
198
- # Mount Gradio interface
199
  app = gr.mount_gradio_app(app, admin_ui, path="/admin-auth")
200
 
201
- # Admin dashboard route
202
  @app.get("/admin")
203
  async def admin_dashboard(email: str = None, password: str = None, response: Response = None):
204
  logger.debug("Admin dashboard accessed")
 
6
  import requests
7
  import logging
8
  import time
9
+ from typing import Optional
10
 
11
  # Configure logging
12
  logging.basicConfig(level=logging.DEBUG)
 
33
  MAX_TOKEN_RETRIES = 3
34
  TOKEN_RETRY_DELAY = 2 # seconds
35
 
 
36
  class TokenManager:
37
  def __init__(self):
38
+ self.token: Optional[str] = None
39
+ self.last_obtained: float = 0
40
+ self.token_expiry: int = 3600 # 1 hour default expiry
41
 
42
+ def get_token(self, force_refresh: bool = False) -> str:
43
  current_time = time.time()
44
 
45
  if not force_refresh and self.token and (current_time - self.last_obtained) < self.token_expiry:
 
54
  }
55
 
56
  logger.debug(f"Attempting to obtain admin token (attempt {attempt + 1})")
57
+
58
+ # Ensure proper headers for JSON content
59
+ headers = {
60
+ "Content-Type": "application/json",
61
+ "Accept": "application/json"
62
+ }
63
+
64
  res = requests.post(
65
  f"{BACKEND_URL}/auth/login",
66
+ json=login_payload, # Using json parameter automatically sets content-type
67
+ headers=headers,
68
  timeout=10
69
  )
70
 
71
+ logger.debug(f"Response status: {res.status_code}, content: {res.text}")
72
+
73
  if res.status_code == 200:
74
  token_data = res.json()
75
  self.token = token_data.get("access_token")
 
77
  raise Exception("No access_token in response")
78
 
79
  self.last_obtained = current_time
 
80
  self.token_expiry = token_data.get("expires_in", 3600)
81
  logger.info("Successfully obtained admin token")
82
  return self.token
83
 
84
  logger.error(f"Token request failed: {res.status_code} - {res.text}")
85
+ if res.status_code == 422:
86
+ raise Exception(f"Validation error: {res.json().get('detail', 'Unknown validation error')}")
87
  if res.status_code == 401:
88
  raise Exception("Invalid admin credentials")
89
 
90
  except requests.exceptions.RequestException as e:
91
  logger.error(f"Network error during token fetch: {str(e)}")
92
+ if attempt < MAX_TOKEN_RETRIES - 1:
93
+ time.sleep(TOKEN_RETRY_DELAY * (attempt + 1)) # Exponential backoff
94
+ continue
95
+ except Exception as e:
96
+ logger.error(f"Unexpected error during token fetch: {str(e)}")
97
  if attempt < MAX_TOKEN_RETRIES - 1:
98
  time.sleep(TOKEN_RETRY_DELAY)
99
  continue
 
102
 
103
  token_manager = TokenManager()
104
 
 
105
  @app.get("/")
106
  def root():
107
  logger.debug("Root endpoint accessed")
108
  return {"message": "πŸš€ FastAPI with MongoDB + JWT is running."}
109
 
 
110
  @app.post("/login")
111
  async def redirect_login(request: Request):
112
  logger.info("Redirecting /login to /auth/login")
113
  return RedirectResponse(url="/auth/login", status_code=307)
114
 
 
115
  def authenticate_admin(email: str = None, password: str = None):
116
  if email != ADMIN_EMAIL or password != ADMIN_PASSWORD:
117
  logger.warning(f"Failed admin login attempt with email: {email}")
118
  raise HTTPException(status_code=401, detail="Unauthorized: Invalid email or password")
 
119
  logger.info(f"Admin authenticated successfully: {email}")
120
  return True
121
 
122
+ def create_doctor(full_name: str, email: str, matricule: str, password: str, specialty: str) -> str:
 
123
  try:
124
  token = token_manager.get_token()
125
 
 
130
  "password": password,
131
  "specialty": specialty,
132
  }
133
+
134
  headers = {
135
  "Authorization": f"Bearer {token}",
136
  "Content-Type": "application/json"
 
158
  if res.status_code == 201:
159
  return "βœ… Doctor created successfully!"
160
 
161
+ error_detail = res.json().get('detail', res.text)
162
  return f"❌ Error: {error_detail} (Status: {res.status_code})"
163
 
164
  except requests.exceptions.RequestException as e:
 
166
  except Exception as e:
167
  return f"❌ System Error: {str(e)}"
168
 
 
169
  admin_ui = gr.Blocks(css="""
170
  .gradio-container {
171
  background-color: #1A1B1F;
 
206
  outputs=output
207
  )
208
 
 
209
  app = gr.mount_gradio_app(app, admin_ui, path="/admin-auth")
210
 
 
211
  @app.get("/admin")
212
  async def admin_dashboard(email: str = None, password: str = None, response: Response = None):
213
  logger.debug("Admin dashboard accessed")