Files
plantillas-proyectos/backend/api/v1/modules/a76/user_tenant/routes.py

148 lines
4.0 KiB
Python

"""
Rutas para gestión de relaciones usuario-tenant
"""
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
from core.database import get_core_db
from core.security import get_current_user
from .service import UserTenantService
from .dto import (
AddUserToTenantRequestDTO,
RemoveUserFromTenantRequestDTO,
UpdateUserRoleRequestDTO,
UserTenantResponseDTO,
UserTenantsResponseDTO,
TenantBasicInfoDTO
)
router = APIRouter(
prefix="/user-tenants",
tags=["User-Tenant Relations"]
)
@router.post("/add", response_model=UserTenantResponseDTO)
def add_user_to_tenant(
data: AddUserToTenantRequestDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Agrega un usuario a un tenant
Requiere permisos de administrador
"""
service = UserTenantService(db)
result = service.add_user_to_tenant(
keycloak_user_id=data.keycloak_user_id,
tenant_id=data.tenant_id,
role=data.role
)
return result
@router.post("/remove")
def remove_user_from_tenant(
data: RemoveUserFromTenantRequestDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Elimina un usuario de un tenant
Requiere permisos de administrador
"""
service = UserTenantService(db)
service.remove_user_from_tenant(
keycloak_user_id=data.keycloak_user_id,
tenant_id=data.tenant_id,
soft_delete=data.soft_delete
)
return {"message": "User removed from tenant successfully"}
@router.put("/update-role", response_model=UserTenantResponseDTO)
def update_user_role(
data: UpdateUserRoleRequestDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Actualiza el rol de un usuario en un tenant
Requiere permisos de administrador
"""
service = UserTenantService(db)
result = service.update_user_role_in_tenant(
keycloak_user_id=data.keycloak_user_id,
tenant_id=data.tenant_id,
role=data.role
)
return result
@router.get("/user/{keycloak_user_id}", response_model=UserTenantsResponseDTO)
def get_user_tenants(
keycloak_user_id: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Obtiene todos los tenants a los que tiene acceso un usuario
Los usuarios solo pueden ver sus propios tenants, a menos que sean admin
"""
# Verificar que el usuario solo pueda ver sus propios tenants (excepto admin)
if current_user.get("sub") != keycloak_user_id:
# TODO: Verificar si es admin
raise HTTPException(
status_code=403,
detail="You can only view your own tenants"
)
service = UserTenantService(db)
tenants = service.get_user_tenants(keycloak_user_id)
return UserTenantsResponseDTO(
keycloak_user_id=keycloak_user_id,
tenants=[TenantBasicInfoDTO.model_validate(t) for t in tenants]
)
@router.get("/tenant/{tenant_id}", response_model=List[UserTenantResponseDTO])
def get_tenant_users(
tenant_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Obtiene todos los usuarios que tienen acceso a un tenant
Requiere permisos de administrador del tenant
"""
service = UserTenantService(db)
user_tenants = service.get_tenant_users(tenant_id)
return user_tenants
@router.get("/check-access/{keycloak_user_id}/{tenant_id}")
def check_user_access(
keycloak_user_id: str,
tenant_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user)
):
"""
Verifica si un usuario tiene acceso a un tenant
"""
service = UserTenantService(db)
has_access = service.user_has_access_to_tenant(keycloak_user_id, tenant_id)
return {
"keycloak_user_id": keycloak_user_id,
"tenant_id": tenant_id,
"has_access": has_access
}