# from fastapi import FastAPI, HTTPException, Request from fastapi.openapi.utils import get_openapi from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError import joblib import pandas as pd import json from App.models import ClientFeatures import time import uuid import json import logging from logging.handlers import RotatingFileHandler from starlette.exceptions import HTTPException as StarletteHTTPException from fastapi.openapi.utils import get_openapi import psutil from typing import List import os app = FastAPI() @app.get("/") def root(): return {"status": "ok"} # Détection Hugging Face RUNNING_IN_HF = "SPACE_ID" in os.environ if RUNNING_IN_HF: LOG_DIR = "/code/logs" else: LOG_DIR = "logs" LOG_FILE = os.path.join(LOG_DIR, "predictions_log.jsonl") # Création du dossier logs dans le conteneur Hugging Face os.makedirs(LOG_DIR, exist_ok=True) # Création du fichier vide s'il n'existe pas if not os.path.exists(LOG_FILE): with open(LOG_FILE, "w") as f: pass logger = logging.getLogger("prediction_logger") # objet qui va enregistrer tous les logs d'informations, warning, erreurs mais de debug logger.setLevel(logging.INFO) # les informations standard de fonctionnement d'API(latence, statut, inputs, outputs, erreurs, etc) : INFO(niveau normal de fonctionnement) handler = RotatingFileHandler( # définit l'endroit où écrire tous les logs "logs/predictions_log.jsonl", maxBytes=5_000_000, # lorsque le fichier dépasse 5Mo, backupCount=3) # renomme le fichier en gardant uniquement les 3 dernières versions; s'il y a un 4ème, le plus ancien est supprimé <> éviter de saturer le disque # Configurer la manière dont les logs seront écrits formatter = logging.Formatter('%(message)s') # les logs écrits doivent être uniquement sous le format "message" <> jsonl sans aucune autre informations supplémentaires handler.setFormatter(formatter) # applique le format definit ci-dessus et écrit dans le fichier logger.addHandler(handler) # connecte le logger au fichier sans quoi rien ne sera écrit # le format jsonl # json: json est un bloc chargé entièrement en memoire mais difficile pour ajouter de nouvelles données sans reécrire tout le fichier; jsonl (json lines) # qui écrit un évènement par ligne sous format json, possibilité d'y ajouter autant de ligne sans modifier les reste et sans recharger tout le fichier # Logs des erreurs 422 @app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): request_id = getattr(request.state, "request_id", str(uuid.uuid4())) log_entry = { "request_id": request_id, "timestamp": time.time(), "path": request.url.path, "method": request.method, "status": "error", "error_message": "ValidationError: " + str(exc.errors()), "latency_ms": None, "request_size_bytes": int(request.headers.get("content-length", 0)), "response_size_bytes": None, "cpu_percent": psutil.cpu_percent(), "ram_percent": psutil.virtual_memory().percent, "system_load": psutil.getloadavg()[0], "num_threads": psutil.Process().num_threads()} logger.info(json.dumps(log_entry)) return JSONResponse( status_code = 422, content={"detail": exc.errors()}) @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request: Request, exc): request_id = getattr(request.state, "request_id", str(uuid.uuid4())) log_entry = { "request_id": request_id, "timestamp": time.time(), "path": request.url.path, "method": request.method, "status": "error", "error_message": f"HTTPException {exc.status_code}: {exc.detail}", "latency_ms": None, "request_size_bytes": int(request.headers.get("content-length", 0)), "response_size_bytes": None, "cpu_percent": psutil.cpu_percent(), "ram_percent": psutil.virtual_memory().percent, "system_load": psutil.getloadavg()[0], "num_threads": psutil.Process().num_threads()} logger.info(json.dumps(log_entry)) return JSONResponse( status_code=exc.status_code, content={"detail": exc.detail}) # Log de chaque requête @app.middleware("http") async def log_requests(request: Request, call_next): start = time.time() # heure de début pour calculer la latence (delai de réponse entre l'envoi de requete et la réponse retournée) request_id = str(uuid.uuid4()) # généré un id pour chaque requete effectuée request.state.request_id = request_id request_size = int(request.headers.get("content-length", 0)) # Taille de la requête try: response = await call_next(request) # exécute la requete et si tout fonctionne renvoie "succès" status = "success" error_message = None except Exception as e: # si çà plante, on récupère l'erreur, la loggues, puis renvoie un message à l'utilisateur status = "error" error_message = str(e) response = JSONResponse( status_code=500, # 500 statut code "côté serveur" # 400 "côté client" content={"detail": "Internal server error"}) latency_ms = (time.time() - start) * 1000 # calcul du temps de traitement en millisecondes (<= 200 <> bon, 200-500 <> acceptable, >= 500 <> bas ou lent) # Taille de la réponse response_size = 0 if hasattr(response, "body"): response_size = len(response.body) log_entry = { # on va enregistrer "request_id": request_id, # l'id de la requete "timestamp": time.time(), # l'heure "path": request.url.path, # l'endpoint appelé "method": request.method, # la méthode "status": status, # le statut (succès/erreur) "error_message": error_message, # le message d'erreur "latency_ms": latency_ms, "request_size_bytes": request_size, "response_size_bytes": response_size, "cpu_percent": psutil.cpu_percent(), "ram_percent": psutil.virtual_memory().percent, "system_load": psutil.getloadavg()[0], "num_threads": psutil.Process().num_threads()} # la latence logger.info(json.dumps(log_entry)) return response # Charger le pipeline MLflow (attend les colonnes ORIGINALES) pipe = joblib.load("./BestModel/pipeline_complet.joblib") # Charger le seuil optimal threshold = joblib.load("./BestModel/best_threshold.joblib") # Charger le mapping clean -> original with open("App/column_mapping.json") as f: COLUMN_MAPPING = json.load(f) @app.post("/predict") def predict(features: ClientFeatures, request: Request): request_id = request.state.request_id # Validation métier data = features.dict() # Âge invalide if "DAYS_BIRTH" in data and data["DAYS_BIRTH"] is not None: # Test attend : âge impossible → 422 + "Âge invalide" if data["DAYS_BIRTH"] > 0 or data["DAYS_BIRTH"] > -18*365: raise HTTPException( status_code=422, detail="Âge invalide") # Revenu invalide if "AMT_INCOME_TOTAL" in data and data["AMT_INCOME_TOTAL"] is not None: if data["AMT_INCOME_TOTAL"] <= 0: raise HTTPException( status_code=422, detail="Revenu invalide") infer_start = time.perf_counter() # Conversion en DataFrame df = pd.DataFrame([data]) # Remapping clean à l'original df = df.rename(columns=COLUMN_MAPPING) # Vérifier que toutes les colonnes attendues par MLflow sont présentes missing = set(pipe.feature_names_in_) - set(df.columns) if missing: raise HTTPException( status_code=422, detail=f"Colonnes manquantes après remapping : {missing}") # Prédiction MLflow score = pipe.predict_proba(df)[0][1] decision = "ACCORDÉ" if score < threshold else "REFUSÉ" infer_time_ms = (time.perf_counter() - infer_start) * 1000 X_monitoring = pipe[:-1].transform(df) X_monitoring = pd.DataFrame(X_monitoring,columns=pipe[:-1].get_feature_names_out()) result = { "score": float(score), "decision": decision, "threshold": float(threshold)} # Log métier log_entry = { "request_id": request_id, "timestamp": time.time(), "inference_ms": infer_time_ms, "path": "/predict", "inputs": X_monitoring.iloc[0].to_dict(), "score": result["score"], "decision": result["decision"], "threshold": result["threshold"], "cpu_percent": psutil.cpu_percent(), "ram_percent": psutil.virtual_memory().percent, "system_load": psutil.getloadavg()[0], "num_threads": psutil.Process().num_threads()} logger.info(json.dumps(log_entry)) return result @app.post("/predict_batch") def predict_batch(clients: List[dict], request: Request): request_id = request.state.request_id # data = [c.dict() for c in clients] df = pd.DataFrame(clients) df = df.rename(columns=COLUMN_MAPPING) missing = set(pipe.feature_names_in_) - set(df.columns) if missing: raise HTTPException( status_code=422, detail=f"Colonnes manquantes après remapping : {missing}") infer_start = time.perf_counter() scores = pipe.predict_proba(df)[:, 1] infer_time_ms = (time.perf_counter() - infer_start) * 1000 decision = ["ACCORDÉ" if s < threshold else "REFUSÉ" for s in scores] X_monitoring = pipe[:-1].transform(df) X_monitoring = pd.DataFrame(X_monitoring,columns=pipe[:-1].get_feature_names_out()) result = [{ "score": float(s), "decision": d, "threshold": float(threshold)} for s, d in zip(scores, decision)] # Log métier log_entry = { "request_id": request_id, "timestamp": time.time(), "inference_ms": infer_time_ms, "path": "/predict_batch", "batch_size": len(df), "inputs": X_monitoring.iloc[0].to_dict(), "cpu_percent": psutil.cpu_percent(), "ram_percent": psutil.virtual_memory().percent, "system_load": psutil.getloadavg()[0], "num_threads": psutil.Process().num_threads()} logger.info(json.dumps(log_entry)) return {"results": result, "inference_ms": infer_time_ms} # Affichage du fichier en live @app.get("/logs") def read_logs(): with open("/code/logs/predictions_log.jsonl") as f: return f.read().splitlines() # Exemple dynamique dans Swagger (/docs) # Charger dataset une seule fois try: df_example = joblib.load("./data/app_test_clean_v2.joblib") except: df_example = None # sécurité si le fichier n'existe pas dans l'environnement def generate_random_example(): if df_example is None: return {} row = df_example.sample(1).iloc[0].to_dict() # Remplacer NaN par None pour Swagger for k, v in row.items(): if isinstance(v, float) and pd.isna(v): row[k] = None return row def custom_openapi(): if app.openapi_schema: return app.openapi_schema schema = get_openapi( title="Home Credit API", version="1.0", routes=app.routes,) # Injecter l'exemple dynamique dans le schéma OpenAPI try: schema["components"]["schemas"]["ClientFeatures"]["example"] = generate_random_example() except KeyError: pass # sécurité si le schéma n'est pas encore généré app.openapi_schema = schema return schema app.openapi = custom_openapi