jlacoste
117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
import os
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from auth import require_admin
|
|
from database import get_db
|
|
from models import Drink, Invitation, User
|
|
from auth import hash_password
|
|
from schemas import AdminResetPassword, AdminToggleAdmin, UserResponse
|
|
|
|
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
|
|
|
|
|
@router.get("/users", response_model=list[UserResponse])
|
|
def list_users(
|
|
admin: User = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
users = db.query(User).order_by(User.created_at.desc()).all()
|
|
return [UserResponse.model_validate(u) for u in users]
|
|
|
|
|
|
@router.delete("/users/{user_id}")
|
|
def delete_user(
|
|
user_id: int,
|
|
admin: User = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
if user_id == admin.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Vous ne pouvez pas vous supprimer vous-même",
|
|
)
|
|
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
uploads_dir = os.path.join(os.path.dirname(__file__), "..", "uploads")
|
|
|
|
drinks = db.query(Drink).filter(Drink.owner_id == user_id).all()
|
|
for drink in drinks:
|
|
if drink.image_path:
|
|
filepath = os.path.join(os.path.dirname(__file__), "..", drink.image_path)
|
|
resolved = os.path.realpath(filepath)
|
|
if resolved.startswith(os.path.realpath(uploads_dir)) and os.path.exists(resolved):
|
|
os.remove(resolved)
|
|
|
|
db.query(Drink).filter(Drink.owner_id == user_id).delete()
|
|
db.query(Invitation).filter(
|
|
(Invitation.created_by == user_id) | (Invitation.used_by == user_id)
|
|
).delete()
|
|
|
|
db.delete(user)
|
|
db.commit()
|
|
return {"detail": "User deleted"}
|
|
|
|
|
|
@router.post("/users/{user_id}/reset-password")
|
|
def reset_user_password(
|
|
user_id: int,
|
|
body: AdminResetPassword,
|
|
admin: User = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
if user_id == admin.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Utilisez /api/auth/change-password pour changer votre propre mot de passe",
|
|
)
|
|
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
user.hashed_password = hash_password(body.new_password)
|
|
user.token_version += 1
|
|
db.commit()
|
|
return {"detail": "Password reset"}
|
|
|
|
|
|
@router.post("/users/{user_id}/toggle-admin")
|
|
def toggle_admin(
|
|
user_id: int,
|
|
body: AdminToggleAdmin,
|
|
admin: User = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
if user_id == admin.id:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Vous ne pouvez pas modifier votre propre statut admin",
|
|
)
|
|
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
user.is_admin = body.is_admin
|
|
user.token_version += 1
|
|
db.commit()
|
|
return {"detail": f"Admin status set to {body.is_admin}"}
|
|
|
|
|
|
@router.get("/stats")
|
|
def stats(
|
|
admin: User = Depends(require_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
return {
|
|
"users": db.query(User).count(),
|
|
"drinks": db.query(Drink).count(),
|
|
"invitations": db.query(Invitation).count(),
|
|
"invitations_used": db.query(Invitation).filter(Invitation.used_by.isnot(None)).count(),
|
|
}
|