Spaces:
Running
Running
#!/usr/bin/env python3 | |
""" | |
Dexcom Sandbox OAuth Integration | |
Complete implementation for Dexcom Sandbox authentication with user selection | |
""" | |
import os | |
import requests | |
import urllib.parse | |
import json | |
import secrets | |
import webbrowser | |
import logging | |
from datetime import datetime, timedelta | |
from typing import Dict, List, Optional, Tuple, Any | |
from dataclasses import dataclass | |
# Setup logging | |
logging.basicConfig(level=logging.INFO) | |
logger = logging.getLogger(__name__) | |
# Load environment variables | |
from dotenv import load_dotenv | |
load_dotenv() | |
# Dexcom Sandbox Configuration | |
CLIENT_ID = os.getenv("DEXCOM_CLIENT_ID", "mLElKHKRwRDVUrAOPBzktFGY7qkTc7Zm") | |
CLIENT_SECRET = os.getenv("DEXCOM_CLIENT_SECRET", "HmFpgyVweuwKrQpf") | |
REDIRECT_URI = "http://localhost:7860/callback" | |
# Dexcom Sandbox Users (selection-based, no passwords) | |
SANDBOX_USERS = { | |
"user1": "SandboxUser1", | |
"user2": "SandboxUser2", | |
"user3": "SandboxUser3", | |
"user4": "SandboxUser4", | |
"user5": "SandboxUser5", | |
"user6": "SandboxUser6 (Dexcom G6)", | |
"user7": "SandboxUser7 (Dexcom G7)", | |
"user8": "SandboxUser8" | |
} | |
# Dexcom API Endpoints - Following Official Documentation | |
# Based on https://developer.dexcom.com/docs/dexcomv2/endpoint-overview/ and v3 | |
# OAuth endpoints are shared between v2 and v3 (always v2 OAuth) | |
OAUTH_ENDPOINTS = { | |
"login": "https://api.dexcom.com/v2/oauth2/login", # Production OAuth login | |
"token": "https://api.dexcom.com/v2/oauth2/token", # Production OAuth token | |
"sandbox_login": "https://developer.dexcom.com/sandbox-login" # Special sandbox login | |
} | |
# API Base URLs per official documentation | |
API_BASE_URLS = { | |
"production": "https://api.dexcom.com", | |
"sandbox": "https://sandbox-api.dexcom.com" | |
} | |
# Recommended configuration (Sandbox login + Production OAuth + Sandbox API) | |
DEFAULT_ENDPOINTS = { | |
"login": "https://developer.dexcom.com/sandbox-login", # Special sandbox login | |
"token": "https://api.dexcom.com/v2/oauth2/token", # Production OAuth (works for sandbox too) | |
"api_v2": "https://sandbox-api.dexcom.com/v2", # Sandbox API v2 | |
"api_v3": "https://sandbox-api.dexcom.com/v3" # Sandbox API v3 | |
} | |
# Alternative configurations for troubleshooting | |
ENDPOINT_CONFIGURATIONS = [ | |
{ | |
"name": "Sandbox Login + Production OAuth + Sandbox API (Recommended)", | |
"login": "https://developer.dexcom.com/sandbox-login", | |
"token": "https://api.dexcom.com/v2/oauth2/token", | |
"api_v2": "https://sandbox-api.dexcom.com/v2", | |
"api_v3": "https://sandbox-api.dexcom.com/v3" | |
}, | |
{ | |
"name": "All Production OAuth + Production API", | |
"login": "https://api.dexcom.com/v2/oauth2/login", | |
"token": "https://api.dexcom.com/v2/oauth2/token", | |
"api_v2": "https://api.dexcom.com/v2", | |
"api_v3": "https://api.dexcom.com/v3" | |
}, | |
{ | |
"name": "All Sandbox (May not work)", | |
"login": "https://sandbox-api.dexcom.com/v2/oauth2/login", | |
"token": "https://sandbox-api.dexcom.com/v2/oauth2/token", | |
"api_v2": "https://sandbox-api.dexcom.com/v2", | |
"api_v3": "https://sandbox-api.dexcom.com/v3" | |
} | |
] | |
class DexcomSandboxUser: | |
"""Profile for Dexcom Sandbox User""" | |
name: str = "Dexcom Sandbox User" | |
age: int = 35 | |
device_type: str = "Dexcom G6 (Sandbox)" | |
username: str = "sandbox_user" | |
password: str = "selection_based" | |
description: str = "Dexcom Sandbox User with OAuth authentication" | |
diabetes_type: str = "Type 1" | |
years_with_diabetes: int = 8 | |
typical_glucose_pattern: str = "sandbox_data" | |
auth_type: str = "dexcom_sandbox" | |
class DexcomSandboxOAuth: | |
""" | |
Dexcom Sandbox OAuth implementation with user selection | |
Flow: | |
1. Generate auth URL β developer.dexcom.com/sandbox-login | |
2. User selects sandbox user from dropdown (no password) | |
3. Get authorization code from callback | |
4. Exchange code for token β sandbox-api.dexcom.com/v2/oauth2/token | |
5. Use token for API calls β sandbox-api.dexcom.com/v2/... | |
""" | |
def __init__(self, api_version: str = "v3"): | |
self.client_id = CLIENT_ID | |
self.client_secret = CLIENT_SECRET | |
self.redirect_uri = REDIRECT_URI | |
self.api_version = api_version # "v2" or "v3" | |
# OAuth state | |
self.access_token = None | |
self.refresh_token = None | |
self.token_expires_at = None | |
self.state = None | |
# Use recommended configuration (sandbox login + production OAuth + sandbox API) | |
self.working_endpoints = DEFAULT_ENDPOINTS.copy() | |
logger.info(f"β Dexcom Sandbox OAuth initialized (API {api_version})") | |
logger.info(f" Client ID: {self.client_id[:8]}...") | |
logger.info(f" Redirect URI: {self.redirect_uri}") | |
if api_version == "v3": | |
logger.info(" Using API v3 for G6, G7, ONE, ONE+ device support") | |
else: | |
logger.info(" Using API v2 for G5, G6 device support") | |
def get_api_base_url(self) -> str: | |
"""Get the correct API base URL for the current version""" | |
if self.api_version == "v3": | |
return self.working_endpoints["api_v3"] | |
else: | |
return self.working_endpoints["api_v2"] | |
def generate_auth_url(self) -> str: | |
"""Generate OAuth authorization URL for Dexcom Sandbox""" | |
# Generate secure state parameter | |
self.state = secrets.token_urlsafe(32) | |
# OAuth parameters | |
params = { | |
'client_id': self.client_id, | |
'redirect_uri': self.redirect_uri, | |
'response_type': 'code', | |
'scope': 'offline_access', | |
'state': self.state | |
} | |
# Build auth URL using current working endpoints | |
query_string = urllib.parse.urlencode(params) | |
auth_url = f"{self.working_endpoints['login']}?{query_string}" | |
logger.info(f"π Generated auth URL: {auth_url}") | |
return auth_url | |
def start_oauth_flow(self) -> Dict[str, Any]: | |
"""Start the Dexcom Sandbox OAuth flow""" | |
try: | |
auth_url = self.generate_auth_url() | |
# Try to open browser automatically | |
try: | |
webbrowser.open(auth_url) | |
browser_opened = True | |
logger.info("β Browser opened automatically") | |
except Exception as e: | |
browser_opened = False | |
logger.warning(f"β οΈ Could not open browser: {e}") | |
return { | |
"success": True, | |
"auth_url": auth_url, | |
"browser_opened": browser_opened, | |
"state": self.state, | |
"instructions": self._get_oauth_instructions(auth_url, browser_opened) | |
} | |
except Exception as e: | |
logger.error(f"Failed to start OAuth flow: {e}") | |
return { | |
"success": False, | |
"error": f"Failed to start OAuth: {str(e)}" | |
} | |
def _get_oauth_instructions(self, auth_url: str, browser_opened: bool) -> str: | |
"""Generate user-friendly OAuth instructions""" | |
browser_status = "β Browser opened automatically" if browser_opened else "β οΈ Please open URL manually" | |
return f""" | |
π **Dexcom Sandbox OAuth Started** | |
{browser_status} | |
**π Step-by-Step Instructions:** | |
1. π Browser should open to: {auth_url} | |
2. π₯ **Select a Sandbox User** from the dropdown: | |
β’ **SandboxUser6** - Dexcom G6 device (recommended) | |
β’ **SandboxUser7** - Dexcom G7 device | |
β’ **SandboxUser1-8** - Various test scenarios | |
3. β Click "Authorize" to grant access | |
4. β **You will get a 404 error - THIS IS EXPECTED!** | |
5. π **Copy the COMPLETE URL** from your browser's address bar | |
6. π₯ Paste the URL below and click "Complete OAuth" | |
**π± Example callback URL:** | |
`http://localhost:7860/callback?code=ABC123XYZ&state=your_state_here` | |
**π― Important:** | |
- No password needed - just select a user and authorize | |
- Copy the **entire URL** (not just the code part) | |
- SandboxUser6 = Dexcom G6 device data (most common) | |
""" | |
def complete_oauth(self, callback_url: str) -> Dict[str, Any]: | |
"""Complete OAuth by processing callback URL and exchanging code for token""" | |
if not callback_url or not callback_url.strip(): | |
return { | |
"success": False, | |
"error": "Please provide the callback URL from your browser" | |
} | |
try: | |
# Extract authorization code from callback URL | |
auth_code = self._extract_auth_code(callback_url) | |
if not auth_code: | |
return { | |
"success": False, | |
"error": "Could not extract authorization code from URL" | |
} | |
# Validate state parameter | |
callback_state = self._extract_state(callback_url) | |
if callback_state != self.state: | |
logger.warning(f"State mismatch: expected {self.state}, got {callback_state}") | |
# Continue anyway for sandbox testing | |
# Exchange code for tokens | |
token_result = self._exchange_code_for_tokens(auth_code) | |
if token_result["success"]: | |
logger.info("β Dexcom Sandbox OAuth completed successfully") | |
return { | |
"success": True, | |
"message": "β Dexcom Sandbox authentication successful", | |
"access_token": self.access_token, | |
"token_expires_at": self.token_expires_at, | |
"user_profile": DexcomSandboxUser() | |
} | |
else: | |
return token_result | |
except Exception as e: | |
logger.error(f"OAuth completion failed: {e}") | |
return { | |
"success": False, | |
"error": f"OAuth completion failed: {str(e)}" | |
} | |
def _extract_auth_code(self, callback_url: str) -> Optional[str]: | |
"""Extract authorization code from callback URL""" | |
try: | |
parsed_url = urllib.parse.urlparse(callback_url) | |
query_params = urllib.parse.parse_qs(parsed_url.query) | |
if 'code' in query_params: | |
code = query_params['code'][0] | |
logger.info(f"β Extracted auth code: {code[:15]}...") | |
return code | |
else: | |
logger.error("No 'code' parameter found in callback URL") | |
return None | |
except Exception as e: | |
logger.error(f"Error extracting auth code: {e}") | |
return None | |
def _extract_state(self, callback_url: str) -> Optional[str]: | |
"""Extract state parameter from callback URL""" | |
try: | |
parsed_url = urllib.parse.urlparse(callback_url) | |
query_params = urllib.parse.parse_qs(parsed_url.query) | |
if 'state' in query_params: | |
return query_params['state'][0] | |
else: | |
return None | |
except Exception as e: | |
logger.error(f"Error extracting state: {e}") | |
return None | |
def _exchange_code_for_tokens(self, auth_code: str) -> Dict[str, Any]: | |
"""Exchange authorization code for access token following Dexcom API v3 guidelines""" | |
# Try sandbox configuration first, then alternatives | |
configurations_to_try = [self.working_endpoints] + ENDPOINT_CONFIGURATIONS | |
for i, endpoint_config in enumerate(configurations_to_try): | |
endpoint_name = endpoint_config.get('name', f'Sandbox Config {i+1}') | |
token_url = endpoint_config["token"] | |
logger.info(f"π Attempting token exchange #{i+1}: {endpoint_name}") | |
logger.info(f" Token URL: {token_url}") | |
# Token exchange data per Dexcom OAuth2 spec | |
data = { | |
'client_id': self.client_id, | |
'client_secret': self.client_secret, | |
'code': auth_code, | |
'grant_type': 'authorization_code', | |
'redirect_uri': self.redirect_uri | |
} | |
# Headers per Dexcom API guidelines | |
headers = { | |
'Content-Type': 'application/x-www-form-urlencoded', | |
'Accept': 'application/json', | |
'User-Agent': 'GlycoAI-DexcomSandbox/1.0', | |
'Cache-Control': 'no-cache' | |
} | |
try: | |
logger.info(f"π€ Auth code: {auth_code[:15]}...") | |
response = requests.post(token_url, data=data, headers=headers, timeout=30) | |
logger.info(f"π¨ Response status: {response.status_code}") | |
logger.info(f"π¨ Response headers: {dict(response.headers)}") | |
if response.status_code == 200: | |
try: | |
token_data = response.json() | |
except json.JSONDecodeError: | |
logger.error(f"β Invalid JSON response: {response.text}") | |
continue | |
# Store tokens | |
self.access_token = token_data.get('access_token') | |
self.refresh_token = token_data.get('refresh_token') | |
if not self.access_token: | |
logger.error(f"β No access_token in response: {token_data}") | |
continue | |
# Calculate expiration | |
expires_in = token_data.get('expires_in', 3600) # Default 1 hour | |
self.token_expires_at = datetime.now() + timedelta(seconds=expires_in) | |
logger.info(f"β Token exchange successful with {endpoint_name}!") | |
logger.info(f" Access token: {self.access_token[:25]}...") | |
logger.info(f" Token type: {token_data.get('token_type', 'Bearer')}") | |
logger.info(f" Expires in: {expires_in} seconds ({expires_in/3600:.1f} hours)") | |
logger.info(f" Scope: {token_data.get('scope', 'offline_access')}") | |
# Update working endpoints to use the successful configuration | |
self.working_endpoints = endpoint_config.copy() | |
return { | |
"success": True, | |
"access_token": self.access_token, | |
"refresh_token": self.refresh_token, | |
"expires_in": expires_in, | |
"working_endpoint": endpoint_config, | |
"token_type": token_data.get('token_type', 'Bearer'), | |
"scope": token_data.get('scope', 'offline_access') | |
} | |
else: | |
# Log detailed error information | |
error_text = response.text | |
try: | |
error_data = response.json() | |
error_message = error_data.get('error_description', | |
error_data.get('error', 'Unknown error')) | |
error_code = error_data.get('error', 'unknown_error') | |
except: | |
error_message = error_text | |
error_code = f"http_{response.status_code}" | |
logger.warning(f"β {endpoint_name} failed: {response.status_code} - {error_message}") | |
# Specific error handling based on Dexcom API behavior | |
if response.status_code == 404: | |
logger.info(f" β Token endpoint not found, trying next configuration...") | |
continue | |
elif response.status_code == 400 and 'invalid_grant' in error_code: | |
logger.info(f" β Invalid/expired authorization code, trying next configuration...") | |
continue | |
elif response.status_code == 401 and 'invalid_client' in error_code: | |
logger.info(f" β Invalid client credentials, trying next configuration...") | |
continue | |
else: | |
logger.info(f" β HTTP {response.status_code} error, trying next configuration...") | |
continue | |
except requests.exceptions.Timeout: | |
logger.warning(f"β±οΈ {endpoint_name} timed out, trying next configuration...") | |
continue | |
except requests.exceptions.RequestException as e: | |
logger.warning(f"π {endpoint_name} network error: {e}, trying next configuration...") | |
continue | |
except Exception as e: | |
logger.warning(f"β {endpoint_name} unexpected error: {e}, trying next configuration...") | |
continue | |
# All endpoints failed | |
logger.error("β All token endpoint configurations failed") | |
return { | |
"success": False, | |
"error": "All token endpoints failed. This may be due to invalid authorization code, expired code, or Dexcom API issues.", | |
"suggestion": "Please try getting a fresh authorization code (they expire in 60 seconds) or check your Dexcom developer credentials.", | |
"endpoints_tried": configurations_to_try, | |
"troubleshooting": [ | |
"1. Make sure you copied the COMPLETE callback URL", | |
"2. Authorization codes expire in 60 seconds - get a fresh one", | |
"3. Verify your CLIENT_ID and CLIENT_SECRET are correct", | |
"4. Check if your app is properly registered in Dexcom developer portal", | |
"5. Ensure redirect URI matches exactly: http://localhost:7860/callback" | |
] | |
} | |
def _ensure_valid_token(self): | |
"""Ensure we have a valid access token""" | |
if not self.access_token: | |
raise Exception("No access token available. Please authenticate first.") | |
if self.token_expires_at and datetime.now() >= self.token_expires_at - timedelta(minutes=5): | |
logger.info("Token expiring soon, need to refresh...") | |
if self.refresh_token: | |
if not self._refresh_token(): | |
raise Exception("Token expired and refresh failed") | |
else: | |
raise Exception("Token expired and no refresh token available") | |
def _refresh_token(self) -> bool: | |
"""Refresh the access token using refresh token""" | |
if not self.refresh_token: | |
return False | |
token_url = self.working_endpoints["token"] | |
data = { | |
'client_id': self.client_id, | |
'client_secret': self.client_secret, | |
'refresh_token': self.refresh_token, | |
'grant_type': 'refresh_token' | |
} | |
headers = { | |
'Content-Type': 'application/x-www-form-urlencoded', | |
'Accept': 'application/json' | |
} | |
try: | |
response = requests.post(token_url, data=data, headers=headers) | |
if response.status_code == 200: | |
token_data = response.json() | |
self.access_token = token_data.get('access_token') | |
if 'refresh_token' in token_data: | |
self.refresh_token = token_data.get('refresh_token') | |
expires_in = token_data.get('expires_in', 7200) | |
self.token_expires_at = datetime.now() + timedelta(seconds=expires_in) | |
logger.info("β Token refreshed successfully") | |
return True | |
else: | |
logger.error(f"Token refresh failed: {response.status_code}") | |
return False | |
except Exception as e: | |
logger.error(f"Token refresh error: {e}") | |
return False | |
def get_auth_headers(self) -> Dict[str, str]: | |
"""Get authorization headers for API calls""" | |
self._ensure_valid_token() | |
return { | |
'Authorization': f'Bearer {self.access_token}', | |
'Accept': 'application/json', | |
'User-Agent': 'GlycoAI-DexcomSandbox/1.0' | |
} | |
def get_data_range(self) -> Dict[str, Any]: | |
"""Get available data range from Dexcom API""" | |
api_base = self.get_api_base_url() | |
url = f"{api_base}/users/self/dataRange" | |
headers = self.get_auth_headers() | |
try: | |
response = requests.get(url, headers=headers, timeout=30) | |
if response.status_code == 200: | |
data = response.json() | |
logger.info(f"β Data range retrieved from API {self.api_version}: {data}") | |
return data | |
else: | |
logger.error(f"Data range API error: {response.status_code} - {response.text}") | |
raise Exception(f"Data range API error: {response.status_code}") | |
except Exception as e: | |
logger.error(f"Error getting data range: {e}") | |
raise | |
def get_glucose_data(self, start_date: str = None, end_date: str = None) -> List[Dict]: | |
"""Get glucose (EGV) data from Dexcom API""" | |
api_base = self.get_api_base_url() | |
url = f"{api_base}/users/self/egvs" | |
headers = self.get_auth_headers() | |
# Per Dexcom documentation: all endpoints except dataRange require startDate and endDate | |
params = {} | |
if start_date: | |
params['startDate'] = start_date | |
if end_date: | |
params['endDate'] = end_date | |
# Validate date range (max 90 days per Dexcom spec) | |
if start_date and end_date: | |
from datetime import datetime | |
start_dt = datetime.fromisoformat(start_date.replace('Z', '+00:00')) | |
end_dt = datetime.fromisoformat(end_date.replace('Z', '+00:00')) | |
delta = (end_dt - start_dt).days | |
if delta > 90: | |
logger.warning(f"Date range {delta} days exceeds Dexcom 90-day limit") | |
# Adjust to last 90 days | |
start_dt = end_dt - timedelta(days=90) | |
params['startDate'] = start_dt.isoformat() + 'Z' | |
logger.info(f"Adjusted to 90-day window: {params['startDate']} to {end_date}") | |
try: | |
logger.info(f"π Fetching glucose data from API {self.api_version}") | |
logger.info(f" URL: {url}") | |
logger.info(f" Params: {params}") | |
response = requests.get(url, headers=headers, params=params, timeout=30) | |
if response.status_code == 200: | |
data = response.json() | |
egvs = data.get('egvs', []) | |
logger.info(f"β Retrieved {len(egvs)} glucose readings from API {self.api_version}") | |
return egvs | |
else: | |
logger.error(f"EGV API error: {response.status_code} - {response.text}") | |
raise Exception(f"EGV API error: {response.status_code}") | |
except Exception as e: | |
logger.error(f"Error getting glucose data: {e}") | |
raise | |
def get_events_data(self, start_date: str = None, end_date: str = None) -> List[Dict]: | |
"""Get events data from Dexcom API""" | |
api_base = self.get_api_base_url() | |
url = f"{api_base}/users/self/events" | |
headers = self.get_auth_headers() | |
params = {} | |
if start_date: | |
params['startDate'] = start_date | |
if end_date: | |
params['endDate'] = end_date | |
try: | |
response = requests.get(url, headers=headers, params=params, timeout=30) | |
if response.status_code == 200: | |
data = response.json() | |
events = data.get('events', []) | |
logger.info(f"β Retrieved {len(events)} events from API {self.api_version}") | |
return events | |
else: | |
logger.warning(f"Events API returned {response.status_code}, continuing without events") | |
return [] | |
except Exception as e: | |
logger.warning(f"Error getting events data: {e}, continuing without events") | |
return [] | |
class DexcomSandboxIntegration: | |
"""Integration wrapper for Gradio app with API version selection""" | |
def __init__(self, api_version: str = "v3"): | |
self.oauth = DexcomSandboxOAuth(api_version=api_version) | |
self.api_version = api_version | |
self.authenticated = False | |
self.user_profile = None | |
self.glucose_data = None | |
self.events_data = None | |
self.data_loaded_at = None | |
def start_oauth(self) -> Tuple[str, bool, bool]: | |
"""Start OAuth flow for Gradio interface""" | |
result = self.oauth.start_oauth_flow() | |
if result["success"]: | |
return ( | |
result["instructions"], | |
True, # Show callback input | |
True # Show complete button | |
) | |
else: | |
return ( | |
f"β Failed to start OAuth: {result['error']}", | |
False, | |
False | |
) | |
def complete_oauth(self, callback_url: str) -> Tuple[str, bool]: | |
"""Complete OAuth flow for Gradio interface""" | |
result = self.oauth.complete_oauth(callback_url) | |
if result["success"]: | |
self.authenticated = True | |
self.user_profile = result["user_profile"] | |
return ( | |
f"β Dexcom Sandbox authenticated successfully! Click 'Load Data' to begin.", | |
True # Show main interface | |
) | |
else: | |
return ( | |
f"β OAuth failed: {result['error']}", | |
False | |
) | |
def load_glucose_data(self, days: int = 14) -> Dict[str, Any]: | |
"""Load glucose data for the specified number of days""" | |
if not self.authenticated: | |
return { | |
"success": False, | |
"error": "Not authenticated. Please complete OAuth first." | |
} | |
try: | |
# Calculate date range | |
end_time = datetime.now() | |
start_time = end_time - timedelta(days=days) | |
# Fetch glucose data | |
self.glucose_data = self.oauth.get_glucose_data( | |
start_date=start_time.isoformat(), | |
end_date=end_time.isoformat() | |
) | |
# Fetch events data | |
self.events_data = self.oauth.get_events_data( | |
start_date=start_time.isoformat(), | |
end_date=end_time.isoformat() | |
) | |
self.data_loaded_at = datetime.now() | |
return { | |
"success": True, | |
"glucose_readings": len(self.glucose_data), | |
"events": len(self.events_data), | |
"date_range": f"{start_time.strftime('%Y-%m-%d')} to {end_time.strftime('%Y-%m-%d')}", | |
"user": self.user_profile.name | |
} | |
except Exception as e: | |
logger.error(f"Failed to load glucose data: {e}") | |
return { | |
"success": False, | |
"error": f"Failed to load data: {str(e)}" | |
} | |
def get_user_profile(self) -> Optional[DexcomSandboxUser]: | |
"""Get authenticated user profile""" | |
return self.user_profile if self.authenticated else None | |
def get_glucose_data_for_ui(self) -> Optional[List[Dict]]: | |
"""Get glucose data formatted for UI display""" | |
return self.glucose_data if self.authenticated else None | |
def get_status(self) -> Dict[str, Any]: | |
"""Get current authentication and data status""" | |
return { | |
"authenticated": self.authenticated, | |
"user": self.user_profile.name if self.user_profile else None, | |
"glucose_readings": len(self.glucose_data) if self.glucose_data else 0, | |
"events": len(self.events_data) if self.events_data else 0, | |
"data_loaded_at": self.data_loaded_at.isoformat() if self.data_loaded_at else None, | |
"token_expires_at": self.oauth.token_expires_at.isoformat() if self.oauth.token_expires_at else None | |
} | |
def debug_oauth_endpoints(): | |
"""Debug function to test all OAuth endpoints""" | |
print("π DEBUGGING DEXCOM OAUTH ENDPOINTS") | |
print("=" * 60) | |
# Test all endpoint configurations | |
all_configs = [DEFAULT_ENDPOINTS] + ENDPOINT_CONFIGURATIONS | |
for i, config in enumerate(all_configs): | |
name = config.get('name', f'Configuration {i+1}') | |
print(f"\nπ§ͺ Testing {name}:") | |
print(f" Login: {config['login']}") | |
print(f" Token: {config['token']}") | |
print(f" API v2: {config['api_v2']}") | |
print(f" API v3: {config['api_v3']}") | |
# Test if endpoints are reachable | |
for endpoint_type, url in config.items(): | |
if endpoint_type == 'name': | |
continue | |
try: | |
# Just check if the endpoint exists (don't send real requests) | |
response = requests.head(url, timeout=5) | |
status = "β Reachable" if response.status_code != 404 else "β 404 Not Found" | |
print(f" {endpoint_type.upper()}: {status} ({response.status_code})") | |
except requests.exceptions.RequestException as e: | |
print(f" {endpoint_type.upper()}: β Error - {e}") | |
print(f"\nπ‘ Key Points from Official Documentation:") | |
print("1. OAuth endpoints are always v2 (even for v3 API calls)") | |
print("2. Sandbox login uses special developer.dexcom.com/sandbox-login") | |
print("3. Token exchange uses production OAuth endpoint for sandbox too") | |
print("4. API v3 supports G6, G7, ONE, ONE+ devices") | |
print("5. API v2 supports G5, G6 devices (legacy)") | |
print("6. Time window max 90 days for all endpoints except dataRange") | |
def test_dexcom_sandbox(): | |
"""Test Dexcom Sandbox OAuth implementation with both API versions""" | |
print("π§ͺ Testing Dexcom Sandbox OAuth Implementation") | |
print("=" * 60) | |
# Run endpoint debug first | |
debug_oauth_endpoints() | |
print(f"\nπ OAuth Flow Test:") | |
# Test both API versions | |
for api_version in ["v3", "v2"]: | |
print(f"\n--- Testing API {api_version} ---") | |
# Initialize OAuth | |
oauth = DexcomSandboxOAuth(api_version=api_version) | |
# Test auth URL generation | |
auth_url = oauth.generate_auth_url() | |
print(f"β Auth URL generated: {auth_url}") | |
# Test integration wrapper | |
integration = DexcomSandboxIntegration(api_version=api_version) | |
instructions, show_input, show_button = integration.start_oauth() | |
print(f"β OAuth flow started for API {api_version}") | |
print(f" Show callback input: {show_input}") | |
print(f" Show complete button: {show_button}") | |
print(f"\nπ Next steps for testing:") | |
print(f"1. Choose API version (v3 recommended for newer devices)") | |
print(f"2. Open auth URL and select sandbox user") | |
print(f"3. SandboxUser6 (G6) works with both v2 and v3") | |
print(f"4. SandboxUser7 (G7) requires API v3") | |
print(f"5. Copy callback URL after 404 error") | |
print(f"6. System will automatically try multiple token endpoints") | |
print(f"\nπ₯ Available Sandbox Users:") | |
for key, name in SANDBOX_USERS.items(): | |
print(f" β’ {name}") | |
if "G6" in name: | |
print(f" β³ π― Works with both API v2 and v3") | |
elif "G7" in name: | |
print(f" β³ π― Requires API v3") | |
return oauth, integration | |
if __name__ == "__main__": | |
test_dexcom_sandbox() |