Spaces:
Sleeping
Sleeping
File size: 3,147 Bytes
62a3a76 |
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 |
import requests
import sys
import os
from sqlalchemy import create_engine, text
from app.core.config import settings
# API Configuration
API_URL = "http://127.0.0.1:8000"
API_KEY = settings.API_KEY
def verify_api():
print("--- Verifying API ---")
# 1. Health Check
try:
resp = requests.get(f"{API_URL}/health")
if resp.status_code == 200:
print("✅ Health check passed")
else:
print(f"❌ Health check failed: {resp.status_code}")
return False
except Exception as e:
print(f"❌ Could not connect to API: {e}")
return False
# 2. Prediction
payload = {
"age": 30,
"genre": "M",
"revenu_mensuel": 5000,
"statut_marital": "Célibataire",
"departement": "R&D",
"poste": "Ingénieur",
"nombre_experiences_precedentes": 2,
"nombre_heures_travailless": 40,
"annee_experience_totale": 5,
"annees_dans_l_entreprise": 2,
"annees_dans_le_poste_actuel": 1,
"satisfaction_employee_environnement": 3,
"note_evaluation_precedente": 3,
"niveau_hierarchique_poste": 2,
"satisfaction_employee_nature_travail": 3,
"satisfaction_employee_equipe": 4,
"satisfaction_employee_equilibre_pro_perso": 3,
"note_evaluation_actuelle": 3,
"heure_supplementaires": "Non",
"augementation_salaire_precedente": "10-15%",
"nombre_participation_pee": 0,
"nb_formations_suivies": 1,
"nombre_employee_sous_responsabilite": 0,
"distance_domicile_travail": 10,
"niveau_education": 3,
"domaine_etude": "Sciences",
"ayant_enfants": "Non",
"frequence_deplacement": "Rare",
"annees_depuis_la_derniere_promotion": 1,
"annes_sous_responsable_actuel": 1
}
headers = {"X-API-KEY": API_KEY}
try:
resp = requests.post(f"{API_URL}/predict", json=payload, headers=headers)
if resp.status_code == 200:
data = resp.json()
print(f"✅ Prediction successful: {data}")
return True
else:
print(f"❌ Prediction failed: {resp.status_code} - {resp.text}")
return False
except Exception as e:
print(f"❌ Prediction request failed: {e}")
return False
def verify_db():
print("\n--- Verifying Database ---")
try:
# Connect to DB
engine = create_engine(settings.DATABASE_URL)
with engine.connect() as conn:
# Check for latest log
result = conn.execute(text("SELECT * FROM prediction_logs ORDER BY id DESC LIMIT 1"))
row = result.fetchone()
if row:
print(f"✅ Database log found: ID={row[0]}, Prediction={row.prediction}")
return True
else:
print("❌ No logs found in database")
return False
except Exception as e:
print(f"❌ Database verification failed: {e}")
return False
if __name__ == "__main__":
if verify_api():
verify_db()
|