Users debug

This commit is contained in:
Fanilo-Nantenaina 2025-12-03 11:00:55 +03:00
parent 39c3397cd4
commit 5ff01c6c45

182
api.py
View file

@ -144,6 +144,32 @@ class BaremeRemiseResponse(BaseModel):
message: str
# À ajouter dans api.py après les imports et avant les endpoints existants
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime
# =====================================================
# MODÈLES PYDANTIC POUR USERS
# =====================================================
class UserResponse(BaseModel):
"""Modèle de réponse pour un utilisateur"""
id: str
email: str
nom: str
prenom: str
role: str
is_verified: bool
is_active: bool
created_at: str
last_login: Optional[str] = None
failed_login_attempts: int = 0
class Config:
from_attributes = True
# =====================================================
# SERVICES EXTERNES (Universign)
# =====================================================
@ -1557,6 +1583,162 @@ async def statut_queue():
}
@app.get("/debug/users", response_model=List[UserResponse], tags=["Debug"])
async def lister_utilisateurs_debug(
session: AsyncSession = Depends(get_session),
limit: int = Query(100, le=1000),
role: Optional[str] = Query(None),
verified_only: bool = Query(False)
):
"""
🔓 **ROUTE DEBUG** - Liste tous les utilisateurs inscrits
**ATTENTION**: Cette route n'est PAS protégée par authentification.
À utiliser uniquement en développement ou à sécuriser en production.
Args:
limit: Nombre maximum d'utilisateurs à retourner
role: Filtrer par rôle (user, admin, commercial)
verified_only: Afficher uniquement les utilisateurs vérifiés
Returns:
Liste des utilisateurs avec leurs informations (mot de passe masqué)
"""
from database import User
from sqlalchemy import select
try:
# Construction de la requête
query = select(User)
# Filtres optionnels
if role:
query = query.where(User.role == role)
if verified_only:
query = query.where(User.is_verified == True)
# Tri par date de création (plus récents en premier)
query = query.order_by(User.created_at.desc()).limit(limit)
# Exécution
result = await session.execute(query)
users = result.scalars().all()
# Conversion en réponse
users_response = []
for user in users:
users_response.append(UserResponse(
id=user.id,
email=user.email,
nom=user.nom,
prenom=user.prenom,
role=user.role,
is_verified=user.is_verified,
is_active=user.is_active,
created_at=user.created_at.isoformat() if user.created_at else "",
last_login=user.last_login.isoformat() if user.last_login else None,
failed_login_attempts=user.failed_login_attempts or 0
))
logger.info(f"📋 Liste utilisateurs retournée: {len(users_response)} résultat(s)")
return users_response
except Exception as e:
logger.error(f"❌ Erreur liste utilisateurs: {e}")
raise HTTPException(500, str(e))
@app.get("/debug/users/stats", tags=["Debug"])
async def statistiques_utilisateurs(
session: AsyncSession = Depends(get_session)
):
"""
📊 **ROUTE DEBUG** - Statistiques sur les utilisateurs
Non protégée - à sécuriser en production
"""
from database import User
from sqlalchemy import select, func
try:
# Total utilisateurs
total_query = select(func.count(User.id))
total_result = await session.execute(total_query)
total = total_result.scalar()
# Utilisateurs vérifiés
verified_query = select(func.count(User.id)).where(User.is_verified == True)
verified_result = await session.execute(verified_query)
verified = verified_result.scalar()
# Utilisateurs actifs
active_query = select(func.count(User.id)).where(User.is_active == True)
active_result = await session.execute(active_query)
active = active_result.scalar()
# Par rôle
roles_query = select(User.role, func.count(User.id)).group_by(User.role)
roles_result = await session.execute(roles_query)
roles_stats = {role: count for role, count in roles_result.all()}
return {
"total_utilisateurs": total,
"utilisateurs_verifies": verified,
"utilisateurs_actifs": active,
"utilisateurs_non_verifies": total - verified,
"repartition_roles": roles_stats,
"taux_verification": f"{(verified/total*100):.1f}%" if total > 0 else "0%"
}
except Exception as e:
logger.error(f"❌ Erreur stats utilisateurs: {e}")
raise HTTPException(500, str(e))
@app.get("/debug/users/{user_id}", response_model=UserResponse, tags=["Debug"])
async def lire_utilisateur_debug(
user_id: str,
session: AsyncSession = Depends(get_session)
):
"""
👤 **ROUTE DEBUG** - Détails d'un utilisateur par ID
Non protégée - à sécuriser en production
"""
from database import User
from sqlalchemy import select
try:
query = select(User).where(User.id == user_id)
result = await session.execute(query)
user = result.scalar_one_or_none()
if not user:
raise HTTPException(404, f"Utilisateur {user_id} introuvable")
return UserResponse(
id=user.id,
email=user.email,
nom=user.nom,
prenom=user.prenom,
role=user.role,
is_verified=user.is_verified,
is_active=user.is_active,
created_at=user.created_at.isoformat() if user.created_at else "",
last_login=user.last_login.isoformat() if user.last_login else None,
failed_login_attempts=user.failed_login_attempts or 0
)
except HTTPException:
raise
except Exception as e:
logger.error(f"❌ Erreur lecture utilisateur: {e}")
raise HTTPException(500, str(e))
# =====================================================
# LANCEMENT
# =====================================================