feat: Implement user management service and frontend integration
- Added UserService to manage Keycloak users with license validation in the backend. - Created API client for user management in the frontend. - Developed user management page with functionalities to create, update, delete, and list users. - Implemented user statistics retrieval and display. - Added dialogs for user creation, editing, password change, and deletion confirmation.
This commit is contained in:
3
backend/api/v1/modules/core/users/__init__.py
Normal file
3
backend/api/v1/modules/core/users/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
"""
|
||||
Módulo de gestión de usuarios (Keycloak)
|
||||
"""
|
||||
89
backend/api/v1/modules/core/users/dto.py
Normal file
89
backend/api/v1/modules/core/users/dto.py
Normal file
@@ -0,0 +1,89 @@
|
||||
"""
|
||||
DTOs para gestión de usuarios de Keycloak
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
|
||||
from pydantic import BaseModel, EmailStr, Field, field_validator
|
||||
|
||||
|
||||
class CreateUserRequestDTO(BaseModel):
|
||||
"""Request para crear un nuevo usuario en Keycloak"""
|
||||
|
||||
email: EmailStr = Field(..., description="Email del usuario")
|
||||
username: str = Field(
|
||||
..., min_length=3, max_length=50, description="Nombre de usuario"
|
||||
)
|
||||
first_name: str = Field(..., min_length=1, max_length=100, description="Nombre")
|
||||
last_name: str = Field(..., min_length=1, max_length=100, description="Apellido")
|
||||
password: str = Field(..., min_length=8, description="Contraseña temporal")
|
||||
role: Optional[str] = Field(None, description="Rol del usuario en el tenant")
|
||||
enabled: bool = Field(True, description="Si el usuario está habilitado")
|
||||
email_verified: bool = Field(False, description="Si el email está verificado")
|
||||
|
||||
|
||||
class UpdateUserRequestDTO(BaseModel):
|
||||
"""Request para actualizar un usuario en Keycloak"""
|
||||
|
||||
first_name: Optional[str] = Field(None, max_length=100)
|
||||
last_name: Optional[str] = Field(None, max_length=100)
|
||||
email: Optional[str] = Field(None, max_length=255)
|
||||
enabled: Optional[bool] = None
|
||||
email_verified: Optional[bool] = None
|
||||
role: Optional[str] = Field(None, description="Rol del usuario en el tenant actual")
|
||||
|
||||
@field_validator("first_name", "last_name", "email")
|
||||
@classmethod
|
||||
def validate_non_empty_string(cls, v: Optional[str]) -> Optional[str]:
|
||||
"""Valida que si el string está presente, no esté vacío"""
|
||||
if v is not None and v.strip() == "":
|
||||
return None # Convertir strings vacíos a None
|
||||
return v
|
||||
|
||||
|
||||
class UserResponseDTO(BaseModel):
|
||||
"""Response con información de usuario de Keycloak"""
|
||||
|
||||
id: str = Field(..., description="ID de Keycloak del usuario")
|
||||
username: str
|
||||
email: str = Field(default="", description="Email del usuario")
|
||||
first_name: str = Field(default="", description="Nombre del usuario")
|
||||
last_name: str = Field(default="", description="Apellido del usuario")
|
||||
enabled: bool
|
||||
email_verified: bool
|
||||
created_timestamp: Optional[int] = None
|
||||
role: Optional[str] = Field(None, description="Rol del usuario en el tenant actual")
|
||||
|
||||
class Config:
|
||||
from_attributes = True
|
||||
|
||||
|
||||
class UserListResponseDTO(BaseModel):
|
||||
"""Response con lista de usuarios"""
|
||||
|
||||
users: List[UserResponseDTO]
|
||||
total: int
|
||||
page: int
|
||||
page_size: int
|
||||
total_pages: int
|
||||
|
||||
|
||||
class ChangePasswordRequestDTO(BaseModel):
|
||||
"""Request para cambiar contraseña de un usuario"""
|
||||
|
||||
password: str = Field(..., min_length=8, description="Nueva contraseña")
|
||||
temporary: bool = Field(
|
||||
True, description="Si es temporal (usuario debe cambiarla al login)"
|
||||
)
|
||||
|
||||
|
||||
class UserStatsDTO(BaseModel):
|
||||
"""Estadísticas de usuarios del tenant"""
|
||||
|
||||
total_users: int
|
||||
active_users: int
|
||||
inactive_users: int
|
||||
max_users_allowed: int
|
||||
users_available: int
|
||||
usage_percentage: float
|
||||
187
backend/api/v1/modules/core/users/routes.py
Normal file
187
backend/api/v1/modules/core/users/routes.py
Normal file
@@ -0,0 +1,187 @@
|
||||
"""
|
||||
Rutas para gestión de usuarios de Keycloak
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user, get_tenant_from_token
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..user_tenant.models import UserTenant
|
||||
from .dto import (
|
||||
ChangePasswordRequestDTO,
|
||||
CreateUserRequestDTO,
|
||||
UpdateUserRequestDTO,
|
||||
UserListResponseDTO,
|
||||
UserResponseDTO,
|
||||
UserStatsDTO,
|
||||
)
|
||||
from .service import UserService
|
||||
|
||||
router = APIRouter(prefix="/users", tags=["Users"])
|
||||
|
||||
|
||||
def get_user_service(
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
) -> UserService:
|
||||
"""Dependency para obtener servicio de usuarios con el tenant y company del usuario actual"""
|
||||
# Obtener keycloak_user_id del usuario actual
|
||||
keycloak_user_id = current_user.get("sub")
|
||||
if not keycloak_user_id:
|
||||
raise HTTPException(status_code=400, detail="User ID not found in token")
|
||||
|
||||
# Buscar el user_tenant activo del usuario
|
||||
user_tenant = (
|
||||
db.query(UserTenant)
|
||||
.filter(
|
||||
UserTenant.keycloak_user_id == keycloak_user_id,
|
||||
UserTenant.is_active == True,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not user_tenant:
|
||||
raise HTTPException(
|
||||
status_code=400, detail="User does not belong to any tenant"
|
||||
)
|
||||
|
||||
return UserService(db, user_tenant.tenant_id, user_tenant.company_id)
|
||||
|
||||
|
||||
@router.get("/stats", response_model=UserStatsDTO)
|
||||
def get_user_statistics(
|
||||
service: UserService = Depends(get_user_service),
|
||||
):
|
||||
"""
|
||||
Obtiene estadísticas de usuarios del tenant actual
|
||||
|
||||
Muestra:
|
||||
- Total de usuarios
|
||||
- Usuarios activos e inactivos
|
||||
- Límite de licencia
|
||||
- Usuarios disponibles
|
||||
- Porcentaje de uso
|
||||
"""
|
||||
return service.get_user_stats()
|
||||
|
||||
|
||||
@router.get("/", response_model=UserListResponseDTO)
|
||||
def list_users(
|
||||
page: int = Query(1, ge=1, description="Número de página"),
|
||||
page_size: int = Query(20, ge=1, le=100, description="Tamaño de página"),
|
||||
search: Optional[str] = Query(None, description="Término de búsqueda"),
|
||||
service: UserService = Depends(get_user_service),
|
||||
):
|
||||
"""
|
||||
Lista todos los usuarios del tenant con paginación
|
||||
|
||||
Se puede filtrar por término de búsqueda (busca en username, email, nombre)
|
||||
"""
|
||||
result = service.get_tenant_users(page=page, page_size=page_size, search=search)
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/{user_id}", response_model=UserResponseDTO)
|
||||
def get_user(
|
||||
user_id: str,
|
||||
service: UserService = Depends(get_user_service),
|
||||
):
|
||||
"""
|
||||
Obtiene información detallada de un usuario específico
|
||||
|
||||
El usuario debe pertenecer al tenant actual
|
||||
"""
|
||||
return service.get_user(user_id)
|
||||
|
||||
|
||||
@router.post("/", response_model=UserResponseDTO, status_code=201)
|
||||
def create_user(
|
||||
data: CreateUserRequestDTO,
|
||||
service: UserService = Depends(get_user_service),
|
||||
):
|
||||
"""
|
||||
Crea un nuevo usuario en Keycloak y lo asocia al tenant
|
||||
|
||||
Validaciones:
|
||||
- Verifica que no se exceda el límite de usuarios de la licencia
|
||||
- Verifica que el email y username sean únicos
|
||||
- Crea el usuario con contraseña temporal
|
||||
|
||||
Nota: El tenant_id se obtiene automáticamente del servicio (del token del usuario actual)
|
||||
"""
|
||||
user = service.create_user(
|
||||
email=data.email,
|
||||
username=data.username,
|
||||
first_name=data.first_name,
|
||||
last_name=data.last_name,
|
||||
password=data.password,
|
||||
role=data.role,
|
||||
enabled=data.enabled,
|
||||
email_verified=data.email_verified,
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
@router.put("/{user_id}", response_model=UserResponseDTO)
|
||||
def update_user(
|
||||
user_id: str,
|
||||
data: UpdateUserRequestDTO,
|
||||
service: UserService = Depends(get_user_service),
|
||||
):
|
||||
"""
|
||||
Actualiza información de un usuario
|
||||
|
||||
Puede actualizar:
|
||||
- Datos personales (nombre, apellido, email)
|
||||
- Estado (habilitado/deshabilitado)
|
||||
- Verificación de email
|
||||
- Rol en el tenant
|
||||
"""
|
||||
user = service.update_user(
|
||||
user_id=user_id,
|
||||
first_name=data.first_name,
|
||||
last_name=data.last_name,
|
||||
email=data.email,
|
||||
enabled=data.enabled,
|
||||
email_verified=data.email_verified,
|
||||
role=data.role,
|
||||
)
|
||||
return user
|
||||
|
||||
|
||||
@router.delete("/{user_id}")
|
||||
def delete_user(
|
||||
user_id: str,
|
||||
soft_delete: bool = Query(
|
||||
True,
|
||||
description="Si es True, solo desactiva. Si es False, elimina permanentemente",
|
||||
),
|
||||
service: UserService = Depends(get_user_service),
|
||||
):
|
||||
"""
|
||||
Elimina un usuario del tenant
|
||||
|
||||
- soft_delete=True: Solo desactiva la relación (recomendado)
|
||||
- soft_delete=False: Elimina permanentemente de Keycloak
|
||||
"""
|
||||
service.delete_user(user_id, soft_delete=soft_delete)
|
||||
return {"message": "User deleted successfully"}
|
||||
|
||||
|
||||
@router.post("/{user_id}/change-password")
|
||||
def change_user_password(
|
||||
user_id: str,
|
||||
data: ChangePasswordRequestDTO,
|
||||
service: UserService = Depends(get_user_service),
|
||||
):
|
||||
"""
|
||||
Cambia la contraseña de un usuario
|
||||
|
||||
- temporary=True: Usuario debe cambiar la contraseña en el próximo login
|
||||
- temporary=False: Contraseña permanente
|
||||
"""
|
||||
service.change_password(user_id, data.password, data.temporary)
|
||||
return {"message": "Password changed successfully"}
|
||||
497
backend/api/v1/modules/core/users/service.py
Normal file
497
backend/api/v1/modules/core/users/service.py
Normal file
@@ -0,0 +1,497 @@
|
||||
"""
|
||||
Servicio para gestionar usuarios de Keycloak con validación de licencias
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from fastapi import HTTPException
|
||||
from keycloak import KeycloakAdmin, KeycloakError
|
||||
from sqlalchemy import and_, func
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from core.config import settings
|
||||
|
||||
from ..licenses.models import License, LicenseStatus
|
||||
from ..user_tenant.models import UserTenant
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _normalize_keycloak_user(
|
||||
user_data: Dict[str, Any], role: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Normaliza los datos de usuario de Keycloak al formato esperado por el DTO
|
||||
|
||||
Keycloak usa camelCase, nuestro DTO usa snake_case
|
||||
"""
|
||||
return {
|
||||
"id": user_data.get("id"),
|
||||
"username": user_data.get("username", ""),
|
||||
"email": user_data.get("email", ""),
|
||||
"first_name": user_data.get("firstName", ""),
|
||||
"last_name": user_data.get("lastName", ""),
|
||||
"enabled": user_data.get("enabled", False),
|
||||
"email_verified": user_data.get("emailVerified", False),
|
||||
"created_timestamp": user_data.get("createdTimestamp"),
|
||||
"role": role,
|
||||
}
|
||||
|
||||
|
||||
class UserService:
|
||||
"""Servicio para gestionar usuarios en Keycloak"""
|
||||
|
||||
def __init__(self, db: Session, tenant_id: int, company_id: int = None):
|
||||
self.db = db
|
||||
self.tenant_id = tenant_id
|
||||
self.company_id = company_id
|
||||
|
||||
# Inicializar cliente admin de Keycloak
|
||||
self.keycloak_admin = KeycloakAdmin(
|
||||
server_url=f"{settings.KEYCLOAK_SERVER_URL}/kcauth",
|
||||
username=settings.KEYCLOAK_ADMIN_USERNAME,
|
||||
password=settings.KEYCLOAK_ADMIN_PASSWORD,
|
||||
realm_name=settings.KEYCLOAK_REALM,
|
||||
verify=True,
|
||||
)
|
||||
|
||||
def _get_license(self) -> License:
|
||||
"""Obtiene la licencia del tenant actual"""
|
||||
license = (
|
||||
self.db.query(License).filter(License.tenant_id == self.tenant_id).first()
|
||||
)
|
||||
|
||||
if not license:
|
||||
raise HTTPException(
|
||||
status_code=404, detail="License not found for this tenant"
|
||||
)
|
||||
|
||||
if license.status != LicenseStatus.ACTIVE:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail=f"License is not active. Current status: {license.status.value}",
|
||||
)
|
||||
|
||||
# Verificar si la licencia está vigente
|
||||
now = datetime.now(license.expires_at.tzinfo)
|
||||
if license.expires_at < now:
|
||||
raise HTTPException(status_code=403, detail="License has expired")
|
||||
|
||||
return license
|
||||
|
||||
def _check_user_limit(self) -> None:
|
||||
"""Verifica si se puede crear un nuevo usuario según la licencia"""
|
||||
license = self._get_license()
|
||||
|
||||
# Contar usuarios activos del tenant
|
||||
active_users = (
|
||||
self.db.query(func.count(UserTenant.id))
|
||||
.filter(
|
||||
and_(
|
||||
UserTenant.tenant_id == self.tenant_id,
|
||||
UserTenant.is_active == True,
|
||||
)
|
||||
)
|
||||
.scalar()
|
||||
)
|
||||
|
||||
if active_users >= license.max_users:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail=f"User limit reached. Your license allows {license.max_users} users. "
|
||||
f"Currently active: {active_users}. Please upgrade your license.",
|
||||
)
|
||||
|
||||
def create_user(
|
||||
self,
|
||||
email: str,
|
||||
username: str,
|
||||
first_name: str,
|
||||
last_name: str,
|
||||
password: str,
|
||||
role: Optional[str] = None,
|
||||
enabled: bool = True,
|
||||
email_verified: bool = False,
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Crea un nuevo usuario en Keycloak y lo asocia al tenant
|
||||
|
||||
Args:
|
||||
email: Email del usuario
|
||||
username: Nombre de usuario
|
||||
first_name: Nombre
|
||||
last_name: Apellido
|
||||
password: Contraseña inicial
|
||||
role: Rol en el tenant
|
||||
enabled: Si el usuario está habilitado
|
||||
email_verified: Si el email está verificado
|
||||
|
||||
Returns:
|
||||
Información del usuario creado
|
||||
|
||||
Raises:
|
||||
HTTPException: Si se alcanza el límite de usuarios o falla la creación
|
||||
"""
|
||||
# Verificar límite de usuarios
|
||||
self._check_user_limit()
|
||||
|
||||
try:
|
||||
# Crear usuario en Keycloak
|
||||
new_user = {
|
||||
"email": email,
|
||||
"username": username,
|
||||
"enabled": enabled,
|
||||
"emailVerified": email_verified,
|
||||
"firstName": first_name,
|
||||
"lastName": last_name,
|
||||
"attributes": {
|
||||
"tenant_id": [
|
||||
str(self.tenant_id)
|
||||
] # Atributo requerido por Keycloak
|
||||
},
|
||||
"credentials": [
|
||||
{
|
||||
"type": "password",
|
||||
"value": password,
|
||||
"temporary": True, # Usuario debe cambiar en primer login
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
user_id = self.keycloak_admin.create_user(new_user)
|
||||
logger.info(f"User created in Keycloak: {user_id}")
|
||||
|
||||
# Obtener company_id si no se proporcionó
|
||||
if not self.company_id:
|
||||
# Obtener la primera company del tenant
|
||||
from api.v1.modules.a76.general_catalogs.company.models import Company
|
||||
|
||||
company = (
|
||||
self.db.query(Company)
|
||||
.filter(Company.tenant_id == self.tenant_id)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not company:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="No company found for this tenant. Please create a company first.",
|
||||
)
|
||||
company_id = company.id
|
||||
else:
|
||||
company_id = self.company_id
|
||||
|
||||
# Crear relación con el tenant
|
||||
user_tenant = UserTenant(
|
||||
keycloak_user_id=user_id,
|
||||
tenant_id=self.tenant_id,
|
||||
company_id=company_id,
|
||||
role=role,
|
||||
is_active=True,
|
||||
)
|
||||
self.db.add(user_tenant)
|
||||
self.db.commit()
|
||||
|
||||
# Obtener información completa del usuario
|
||||
user_info = self.keycloak_admin.get_user(user_id)
|
||||
|
||||
return _normalize_keycloak_user(user_info, role)
|
||||
|
||||
except KeycloakError as e:
|
||||
logger.error(f"Keycloak error creating user: {str(e)}")
|
||||
self.db.rollback()
|
||||
|
||||
# Manejar errores específicos
|
||||
if "User exists with same email" in str(e):
|
||||
raise HTTPException(
|
||||
status_code=409, detail="A user with this email already exists"
|
||||
)
|
||||
elif "User exists with same username" in str(e):
|
||||
raise HTTPException(
|
||||
status_code=409, detail="A user with this username already exists"
|
||||
)
|
||||
|
||||
raise HTTPException(
|
||||
status_code=500, detail=f"Error creating user in Keycloak: {str(e)}"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Unexpected error creating user: {str(e)}")
|
||||
self.db.rollback()
|
||||
raise HTTPException(
|
||||
status_code=500, detail=f"Error creating user: {str(e)}"
|
||||
)
|
||||
|
||||
def get_tenant_users(
|
||||
self, page: int = 1, page_size: int = 20, search: Optional[str] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Obtiene todos los usuarios del tenant con paginación
|
||||
|
||||
Args:
|
||||
page: Número de página (1-indexed)
|
||||
page_size: Tamaño de página
|
||||
search: Término de búsqueda (busca en username, email, nombre)
|
||||
|
||||
Returns:
|
||||
Dict con usuarios y metadatos de paginación
|
||||
"""
|
||||
try:
|
||||
# Obtener relaciones usuario-tenant
|
||||
query = self.db.query(UserTenant).filter(
|
||||
and_(
|
||||
UserTenant.tenant_id == self.tenant_id,
|
||||
UserTenant.is_active == True,
|
||||
)
|
||||
)
|
||||
|
||||
total = query.count()
|
||||
|
||||
# Calcular offset
|
||||
offset = (page - 1) * page_size
|
||||
user_tenants = query.offset(offset).limit(page_size).all()
|
||||
|
||||
# Obtener información de Keycloak para cada usuario
|
||||
users = []
|
||||
for ut in user_tenants:
|
||||
try:
|
||||
user_info = self.keycloak_admin.get_user(ut.keycloak_user_id)
|
||||
normalized_user = _normalize_keycloak_user(user_info, ut.role)
|
||||
|
||||
# Filtrar por búsqueda si se proporciona
|
||||
if search:
|
||||
search_lower = search.lower()
|
||||
if (
|
||||
search_lower in normalized_user.get("username", "").lower()
|
||||
or search_lower in normalized_user.get("email", "").lower()
|
||||
or search_lower
|
||||
in normalized_user.get("first_name", "").lower()
|
||||
or search_lower
|
||||
in normalized_user.get("last_name", "").lower()
|
||||
):
|
||||
users.append(normalized_user)
|
||||
else:
|
||||
users.append(normalized_user)
|
||||
|
||||
except KeycloakError as e:
|
||||
logger.warning(
|
||||
f"Could not fetch user {ut.keycloak_user_id} from Keycloak: {str(e)}"
|
||||
)
|
||||
continue
|
||||
|
||||
total_pages = (total + page_size - 1) // page_size
|
||||
|
||||
return {
|
||||
"users": users,
|
||||
"total": len(users) if search else total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"total_pages": total_pages,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting tenant users: {str(e)}")
|
||||
raise HTTPException(
|
||||
status_code=500, detail=f"Error getting users: {str(e)}"
|
||||
)
|
||||
|
||||
def get_user(self, user_id: str) -> Dict[str, Any]:
|
||||
"""Obtiene un usuario específico del tenant"""
|
||||
# Verificar que el usuario pertenece al tenant
|
||||
user_tenant = (
|
||||
self.db.query(UserTenant)
|
||||
.filter(
|
||||
and_(
|
||||
UserTenant.keycloak_user_id == user_id,
|
||||
UserTenant.tenant_id == self.tenant_id,
|
||||
UserTenant.is_active == True,
|
||||
)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not user_tenant:
|
||||
raise HTTPException(status_code=404, detail="User not found in this tenant")
|
||||
|
||||
try:
|
||||
user_info = self.keycloak_admin.get_user(user_id)
|
||||
return _normalize_keycloak_user(user_info, user_tenant.role)
|
||||
except KeycloakError as e:
|
||||
logger.error(f"Error getting user from Keycloak: {str(e)}")
|
||||
raise HTTPException(status_code=404, detail="User not found in Keycloak")
|
||||
|
||||
def update_user(
|
||||
self,
|
||||
user_id: str,
|
||||
first_name: Optional[str] = None,
|
||||
last_name: Optional[str] = None,
|
||||
email: Optional[str] = None,
|
||||
enabled: Optional[bool] = None,
|
||||
email_verified: Optional[bool] = None,
|
||||
role: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Actualiza información de un usuario"""
|
||||
# Verificar que el usuario pertenece al tenant
|
||||
user_tenant = (
|
||||
self.db.query(UserTenant)
|
||||
.filter(
|
||||
and_(
|
||||
UserTenant.keycloak_user_id == user_id,
|
||||
UserTenant.tenant_id == self.tenant_id,
|
||||
)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not user_tenant:
|
||||
raise HTTPException(status_code=404, detail="User not found in this tenant")
|
||||
|
||||
try:
|
||||
# Preparar datos de actualización para Keycloak
|
||||
update_data = {}
|
||||
if first_name is not None:
|
||||
update_data["firstName"] = first_name
|
||||
if last_name is not None:
|
||||
update_data["lastName"] = last_name
|
||||
if email is not None:
|
||||
update_data["email"] = email
|
||||
if enabled is not None:
|
||||
update_data["enabled"] = enabled
|
||||
if email_verified is not None:
|
||||
update_data["emailVerified"] = email_verified
|
||||
|
||||
# Actualizar en Keycloak si hay cambios
|
||||
if update_data:
|
||||
self.keycloak_admin.update_user(user_id, update_data)
|
||||
|
||||
# Actualizar rol en UserTenant si se proporciona
|
||||
if role is not None:
|
||||
user_tenant.role = role
|
||||
self.db.commit()
|
||||
|
||||
# Obtener información actualizada
|
||||
user_info = self.keycloak_admin.get_user(user_id)
|
||||
|
||||
return _normalize_keycloak_user(user_info, user_tenant.role)
|
||||
|
||||
except KeycloakError as e:
|
||||
logger.error(f"Error updating user in Keycloak: {str(e)}")
|
||||
self.db.rollback()
|
||||
raise HTTPException(
|
||||
status_code=500, detail=f"Error updating user: {str(e)}"
|
||||
)
|
||||
|
||||
def delete_user(self, user_id: str, soft_delete: bool = True) -> None:
|
||||
"""
|
||||
Elimina un usuario del tenant
|
||||
|
||||
Args:
|
||||
user_id: ID del usuario en Keycloak
|
||||
soft_delete: Si es True, solo desactiva. Si es False, elimina de Keycloak
|
||||
"""
|
||||
# Verificar que el usuario pertenece al tenant
|
||||
user_tenant = (
|
||||
self.db.query(UserTenant)
|
||||
.filter(
|
||||
and_(
|
||||
UserTenant.keycloak_user_id == user_id,
|
||||
UserTenant.tenant_id == self.tenant_id,
|
||||
)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not user_tenant:
|
||||
raise HTTPException(status_code=404, detail="User not found in this tenant")
|
||||
|
||||
try:
|
||||
if soft_delete:
|
||||
# Solo desactivar la relación
|
||||
user_tenant.is_active = False
|
||||
self.db.commit()
|
||||
else:
|
||||
# Eliminar permanentemente de Keycloak
|
||||
self.keycloak_admin.delete_user(user_id)
|
||||
# Eliminar relación
|
||||
self.db.delete(user_tenant)
|
||||
self.db.commit()
|
||||
|
||||
except KeycloakError as e:
|
||||
logger.error(f"Error deleting user from Keycloak: {str(e)}")
|
||||
self.db.rollback()
|
||||
raise HTTPException(
|
||||
status_code=500, detail=f"Error deleting user: {str(e)}"
|
||||
)
|
||||
|
||||
def change_password(
|
||||
self, user_id: str, password: str, temporary: bool = True
|
||||
) -> None:
|
||||
"""Cambia la contraseña de un usuario"""
|
||||
# Verificar que el usuario pertenece al tenant
|
||||
user_tenant = (
|
||||
self.db.query(UserTenant)
|
||||
.filter(
|
||||
and_(
|
||||
UserTenant.keycloak_user_id == user_id,
|
||||
UserTenant.tenant_id == self.tenant_id,
|
||||
UserTenant.is_active == True,
|
||||
)
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not user_tenant:
|
||||
raise HTTPException(status_code=404, detail="User not found in this tenant")
|
||||
|
||||
try:
|
||||
self.keycloak_admin.set_user_password(
|
||||
user_id, password, temporary=temporary
|
||||
)
|
||||
except KeycloakError as e:
|
||||
logger.error(f"Error changing user password: {str(e)}")
|
||||
raise HTTPException(
|
||||
status_code=500, detail=f"Error changing password: {str(e)}"
|
||||
)
|
||||
|
||||
def get_user_stats(self) -> Dict[str, Any]:
|
||||
"""Obtiene estadísticas de usuarios del tenant"""
|
||||
license = self._get_license()
|
||||
|
||||
# Contar usuarios activos e inactivos
|
||||
active_users = (
|
||||
self.db.query(func.count(UserTenant.id))
|
||||
.filter(
|
||||
and_(
|
||||
UserTenant.tenant_id == self.tenant_id,
|
||||
UserTenant.is_active == True,
|
||||
)
|
||||
)
|
||||
.scalar()
|
||||
)
|
||||
|
||||
inactive_users = (
|
||||
self.db.query(func.count(UserTenant.id))
|
||||
.filter(
|
||||
and_(
|
||||
UserTenant.tenant_id == self.tenant_id,
|
||||
UserTenant.is_active == False,
|
||||
)
|
||||
)
|
||||
.scalar()
|
||||
)
|
||||
|
||||
total_users = active_users + inactive_users
|
||||
users_available = max(0, license.max_users - active_users)
|
||||
usage_percentage = (
|
||||
(active_users / license.max_users * 100) if license.max_users > 0 else 0
|
||||
)
|
||||
|
||||
return {
|
||||
"total_users": total_users,
|
||||
"active_users": active_users,
|
||||
"inactive_users": inactive_users,
|
||||
"max_users_allowed": license.max_users,
|
||||
"users_available": users_available,
|
||||
"usage_percentage": round(usage_percentage, 2),
|
||||
}
|
||||
Reference in New Issue
Block a user