101 lines
3.5 KiB
Plaintext
101 lines
3.5 KiB
Plaintext
# Endpoints para gestión de roles - Agregar después de los endpoints de usuarios
|
|
|
|
# ============= ROLE ENDPOINTS =============
|
|
@app.get("/api/roles", response_model=List[schemas.Role])
|
|
def get_roles(
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Lista todos los roles disponibles (cualquier usuario autenticado)"""
|
|
return db.query(models.Role).all()
|
|
|
|
@app.get("/api/roles/{role_id}", response_model=schemas.Role)
|
|
def get_role(
|
|
role_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Obtiene un rol específico"""
|
|
require_permission(current_user, "can_manage_roles")
|
|
|
|
role = db.query(models.Role).filter(models.Role.id == role_id).first()
|
|
if not role:
|
|
raise HTTPException(status_code=404, detail="Rol no encontrado")
|
|
return role
|
|
|
|
@app.post("/api/roles", response_model=schemas.Role)
|
|
def create_role(
|
|
role: schemas.RoleCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Crea un nuevo rol (solo administrador)"""
|
|
require_permission(current_user, "can_manage_roles")
|
|
|
|
# Verificar si el rol ya existe
|
|
existing = db.query(models.Role).filter(models.Role.name == role.name).first()
|
|
if existing:
|
|
raise HTTPException(status_code=400, detail="El rol ya existe")
|
|
|
|
db_role = models.Role(**role.dict())
|
|
db.add(db_role)
|
|
db.commit()
|
|
db.refresh(db_role)
|
|
return db_role
|
|
|
|
@app.put("/api/roles/{role_id}", response_model=schemas.Role)
|
|
def update_role(
|
|
role_id: int,
|
|
role_update: schemas.RoleUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Actualiza un rol existente (solo administrador)"""
|
|
require_permission(current_user, "can_manage_roles")
|
|
|
|
db_role = db.query(models.Role).filter(models.Role.id == role_id).first()
|
|
if not db_role:
|
|
raise HTTPException(status_code=404, detail="Rol no encontrado")
|
|
|
|
# No permitir editar roles predefinidos (1, 2, 3)
|
|
if role_id in [1, 2, 3]:
|
|
raise HTTPException(status_code=403, detail="No se pueden editar roles predefinidos")
|
|
|
|
# Actualizar campos
|
|
update_data = role_update.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(db_role, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(db_role)
|
|
return db_role
|
|
|
|
@app.delete("/api/roles/{role_id}")
|
|
def delete_role(
|
|
role_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: models.User = Depends(get_current_user)
|
|
):
|
|
"""Elimina un rol (solo administrador, no permite eliminar roles predefinidos)"""
|
|
require_permission(current_user, "can_manage_roles")
|
|
|
|
# No permitir eliminar roles predefinidos
|
|
if role_id in [1, 2, 3]:
|
|
raise HTTPException(status_code=403, detail="No se pueden eliminar roles predefinidos")
|
|
|
|
db_role = db.query(models.Role).filter(models.Role.id == role_id).first()
|
|
if not db_role:
|
|
raise HTTPException(status_code=404, detail="Rol no encontrado")
|
|
|
|
# Verificar si hay usuarios con este rol
|
|
users_count = db.query(models.User).filter(models.User.role_id == role_id).count()
|
|
if users_count > 0:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"No se puede eliminar el rol porque tiene {users_count} usuario(s) asignado(s)"
|
|
)
|
|
|
|
db.delete(db_role)
|
|
db.commit()
|
|
return {"message": "Rol eliminado correctamente", "role_id": role_id}
|