File size: 4,904 Bytes
b666236 4c89353 b666236 | 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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 | """
Module de logging des prédictions dans la base de données PostgreSQL.
Si la base de données n'est pas disponible (ex: conteneur Docker sans PostgreSQL),
les fonctions ne lèvent pas d'erreur et laissent un avertissement dans les logs.
"""
import logging
import os
from dotenv import load_dotenv
logger = logging.getLogger(__name__)
# Vérifie si la DB est configurée
load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")
DB_AVAILABLE = DATABASE_URL is not None
def _get_session():
"""
Ouvre une session SQLAlchemy si la DB est disponible.
Returns:
Session SQLAlchemy ou None si la DB n'est pas configurée
"""
if not DB_AVAILABLE:
return None
try:
from src.db.database import SessionLocal
return SessionLocal()
except Exception as e:
logger.warning(f"Impossible de se connecter à la base de données : {e}")
return None
def log_prediction_input(donnees: dict) -> int | None:
"""
Enregistre les données brutes d'un employé dans prediction_inputs.
Args:
donnees: Dict avec les données de l'employé (validées par Pydantic)
Returns:
ID de l'input inséré, ou None si la DB n'est pas disponible
"""
if not DB_AVAILABLE:
logger.info("DB non configurée : log_prediction_input ignoré")
return None
try:
from src.db.models import PredictionInput
except Exception:
logger.warning("Impossible d'importer les modèles DB")
return None
session = _get_session()
if session is None:
return None
try:
prediction_input = PredictionInput(**donnees)
session.add(prediction_input)
session.commit()
session.refresh(prediction_input)
input_id: int = prediction_input.id # type: ignore[assignment]
logger.info(f"Input inséré en DB (id={input_id})")
return input_id
except Exception as e:
session.rollback()
logger.warning(f"Erreur lors de l'insertion input : {e}")
return None
finally:
session.close()
def log_prediction_output(
input_id: int | None, prediction: str, probabilite: float, classe: int
) -> bool:
"""
Enregistre le résultat d'une prédiction dans prediction_outputs.
Args:
input_id: ID de l'input dans prediction_inputs
prediction: "Oui" ou "Non"
probabilite: Probabilité de départ (entre 0 et 1)
classe: 0 ou 1
Returns:
True si l'insertion a réussi, False sinon
"""
if not DB_AVAILABLE or input_id is None:
logger.info(
"DB non configurée ou input_id manquant : log_prediction_output ignoré"
)
return False
try:
from src.db.models import PredictionOutput
except Exception:
logger.warning("Impossible d'importer les modèles DB")
return False
session = _get_session()
if session is None:
return False
try:
prediction_output = PredictionOutput(
input_id=input_id,
prediction=prediction,
probabilite=probabilite,
classe=classe,
)
session.add(prediction_output)
session.commit()
logger.info(f"Output inséré en DB (input_id={input_id})")
return True
except Exception as e:
session.rollback()
logger.warning(f"Erreur lors de l'insertion output : {e}")
return False
finally:
session.close()
def log_api_operation(
operation: str,
table_cible: str,
details: str | None = None,
statut: str = "SUCCESS",
) -> bool:
"""
Enregistre une opération API dans api_logs.
Args:
operation: Type d'opération (ex: "INSERT", "PREDICT")
table_cible: Table concernée (ex: "prediction_inputs")
details: Détails optionnels de l'opération
statut: "SUCCESS" ou "ERROR"
Returns:
True si l'insertion a réussi, False sinon
"""
if not DB_AVAILABLE:
logger.info(
f"DB non configurée : log_api_operation ignoré ({operation} {table_cible})"
)
return False
try:
from src.db.models import ApiLog
except Exception:
logger.warning("Impossible d'importer les modèles DB")
return False
session = _get_session()
if session is None:
return False
try:
api_log = ApiLog(
operation=operation,
table_cible=table_cible,
details=details,
statut=statut,
)
session.add(api_log)
session.commit()
logger.info(f"Log API inséré : {operation} {table_cible} ({statut})")
return True
except Exception as e:
session.rollback()
logger.warning(f"Erreur lors de l'insertion log API : {e}")
return False
finally:
session.close()
|