ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement license and tenant management APIs

Browse Source 
- Added endpoints for license management including creation, retrieval, update, validation, and usage tracking.
- Developed service layer for business logic related to licenses.
- Introduced tenant management APIs for creating, updating, listing, and deleting tenants.
- Implemented user-tenant relationship management with endpoints for adding, removing, and updating user roles in tenants.
- Created DTOs for data transfer between layers and models for ORM mapping.
- Enhanced logging and error handling across services.
This commit is contained in:
acazares
2025-12-09 11:51:18 -06:00
parent 3992a9436b
commit 44eef08445
23 changed files with 25 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
backend/api/v1/modules/a76/auth/__init__.py
View File

@@ -1,7 +0,0 @@
"""
Módulo de Authentication
"""
from .routes import router
__all__ = ["router"]

154
backend/api/v1/modules/a76/auth/dto.py
View File

@@ -1,154 +0,0 @@
"""
DTOs para módulo de autenticación
"""
from typing import Optional
from pydantic import BaseModel, EmailStr, Field
class LoginRequestDTO(BaseModel):
"""DTO para solicitud de login"""
username: str = Field(..., description="Usuario o email")
password: str = Field(..., min_length=6, description="Contraseña")
tenant_slug: str = Field(..., description="Slug del tenant")
class Config:
json_schema_extra = {
"example": {
"username": "usuario@ejemplo.com",
"password": "password123",
"tenant_slug": "empresa-abc",
}
}
class TokenResponseDTO(BaseModel):
"""DTO para respuesta de token"""
access_token: str
refresh_token: str
token_type: str = "bearer"
expires_in: int
class Config:
json_schema_extra = {
"example": {
"access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
"refresh_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
"token_type": "bearer",
"expires_in": 3600,
}
}
class RefreshTokenRequestDTO(BaseModel):
"""DTO para solicitud de refresh token"""
refresh_token: str = Field(..., description="Refresh token")
class UserInfoResponseDTO(BaseModel):
"""DTO para información de usuario"""
sub: str
email: Optional[str] = None
name: Optional[str] = None
preferred_username: Optional[str] = None
tenant_id: Optional[int] = None
roles: list[str] = []
class Config:
json_schema_extra = {
"example": {
"sub": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"email": "usuario@ejemplo.com",
"name": "Juan Pérez",
"preferred_username": "jperez",
"tenant_id": 1,
"roles": ["user", "admin"],
}
}
class LogoutRequestDTO(BaseModel):
"""DTO para solicitud de logout"""
refresh_token: str = Field(..., description="Refresh token para invalidar")
class RegisterRequestDTO(BaseModel):
"""DTO para solicitud de registro"""
username: str = Field(
..., min_length=3, max_length=50, description="Nombre de usuario"
)
email: EmailStr = Field(..., description="Email del usuario")
password: str = Field(..., min_length=8, description="Contraseña")
first_name: str = Field(..., min_length=2, max_length=50, description="Nombre")
last_name: str = Field(..., min_length=2, max_length=50, description="Apellido")
tenant_slug: str = Field(..., description="Slug del tenant")
class Config:
json_schema_extra = {
"example": {
"username": "jperez",
"email": "jperez@ejemplo.com",
"password": "MiPassword123!",
"first_name": "Juan",
"last_name": "Pérez",
"tenant_slug": "empresa-abc",
}
}
class RegisterResponseDTO(BaseModel):
"""DTO para respuesta de registro"""
user_id: str
username: str
email: str
message: str
class Config:
json_schema_extra = {
"example": {
"user_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"username": "jperez",
"email": "jperez@ejemplo.com",
"message": "User registered successfully",
}
}
class ExchangeCodeRequestDTO(BaseModel):
"""DTO para intercambiar authorization code por tokens (OAuth2 flow)"""
code: str = Field(..., description="Authorization code de OAuth2")
redirect_uri: str = Field(..., description="Redirect URI usado en la autorización")
tenant_slug: Optional[str] = Field(None, description="Slug del tenant (opcional)")
class Config:
json_schema_extra = {
"example": {
"code": "eyJhbGciOiJkaXIiLCJlbmMiOiJBMTI4Q0JDLUhTMjU2Ii...",
"redirect_uri": "http://localhost:5173/auth/callback",
"tenant_slug": "empresa-abc",
}
}
class SetCookieRequestDTO(BaseModel):
"""DTO para establecer cookies de autenticación"""
access_token: str = Field(..., description="Access token JWT")
refresh_token: str = Field(..., description="Refresh token JWT")
class Config:
json_schema_extra = {
"example": {
"access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
"refresh_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
}
}

174
backend/api/v1/modules/a76/auth/routes.py
View File

@@ -1,174 +0,0 @@
"""
Endpoints API para autenticación
"""
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from sqlalchemy.orm import Session
from .dto import (
ExchangeCodeRequestDTO,
LoginRequestDTO,
LogoutRequestDTO,
RefreshTokenRequestDTO,
RegisterRequestDTO,
RegisterResponseDTO,
SetCookieRequestDTO,
TokenResponseDTO,
UserInfoResponseDTO,
)
from .service import AuthService
router = APIRouter(prefix="/auth", tags=["Authentication"])
security = HTTPBearer()
@router.post("/register", response_model=RegisterResponseDTO, status_code=201)
async def register(
register_data: RegisterRequestDTO, db: Session = Depends(get_core_db)
):
"""
Registra un nuevo usuario en Keycloak
El usuario debe proporcionar:
- username: Nombre de usuario único
- email: Email único
- password: Contraseña (mínimo 8 caracteres)
- first_name: Nombre
- last_name: Apellido
- tenant_slug: Slug del tenant al que pertenece
El usuario se crea automáticamente en Keycloak con:
- Cuenta habilitada
- Rol 'user' asignado por defecto
- Atributos de tenant
"""
service = AuthService(db)
return service.register(register_data)
@router.post("/login", response_model=TokenResponseDTO)
async def login(login_data: LoginRequestDTO, db: Session = Depends(get_core_db)):
"""
Autentica usuario con Keycloak y retorna tokens JWT
El usuario debe proporcionar:
- username: Usuario o email
- password: Contraseña
- tenant_slug: Slug del tenant al que pertenece
"""
service = AuthService(db)
return service.login(login_data)
@router.post("/refresh", response_model=TokenResponseDTO)
async def refresh_token(
refresh_data: RefreshTokenRequestDTO, db: Session = Depends(get_core_db)
):
"""
Refresca el access token usando el refresh token
"""
service = AuthService(db)
return service.refresh_token(refresh_data)
@router.get("/me", response_model=UserInfoResponseDTO)
async def get_current_user_info(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: Session = Depends(get_core_db),
):
"""
Obtiene información del usuario actual desde el token
"""
service = AuthService(db)
return service.get_user_info(credentials.credentials)
@router.post("/logout")
async def logout(
logout_data: LogoutRequestDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Cierra sesión invalidando el refresh token
"""
service = AuthService(db)
return service.logout(logout_data)
@router.post("/exchange-code", response_model=TokenResponseDTO)
async def exchange_code(
exchange_data: ExchangeCodeRequestDTO, db: Session = Depends(get_core_db)
):
"""
Intercambia un authorization code de OAuth2 por tokens
Este endpoint es útil cuando el frontend usa el flujo de autorización
con proveedores externos (Microsoft, Google, etc.) a través de Keycloak.
El código se obtiene después de que el usuario se autentica con el proveedor
externo y Keycloak lo redirige al frontend con el código en los query params.
"""
service = AuthService(db)
return service.exchange_code(exchange_data)
@router.post("/set-cookie")
async def set_cookie(
cookie_data: SetCookieRequestDTO,
response: Response,
db: Session = Depends(get_core_db),
):
"""
Establece cookies HttpOnly con los tokens de autenticación
Este endpoint se llama desde el frontend después de una autenticación
SSO exitosa para establecer las cookies de sesión necesarias para
la validación server-side en los layouts protegidos.
Las cookies se configuran como:
- HttpOnly: No accesibles desde JavaScript (mayor seguridad)
- Secure: Solo se envían por HTTPS (en producción)
- SameSite=Lax: Protección contra CSRF
- Max-Age: Tiempo de vida del token
"""
# Validar que los tokens sean válidos decodificándolos
service = AuthService(db)
try:
# Validar el access token
user_info = service.get_user_info(cookie_data.access_token)
# Establecer las cookies
# Access token cookie
response.set_cookie(
key="access_token",
value=cookie_data.access_token,
httponly=True, # No accesible desde JavaScript
secure=False, # TODO: Cambiar a True en producción con HTTPS
samesite="lax", # Protección CSRF
max_age=3600, # 1 hora (ajustar según configuración del token)
path="/",
)
# Refresh token cookie
response.set_cookie(
key="refresh_token",
value=cookie_data.refresh_token,
httponly=True,
secure=False, # TODO: Cambiar a True en producción con HTTPS
samesite="lax",
max_age=86400, # 24 horas (ajustar según configuración del token)
path="/",
)
return {
"success": True,
"message": "Cookies establecidas correctamente",
"user": user_info,
}
except Exception as e:
raise HTTPException(status_code=400, detail=f"Error validando tokens: {str(e)}")

438
backend/api/v1/modules/a76/auth/service.py
View File

@@ -1,438 +0,0 @@
"""
Servicio de autenticación con Keycloak
"""
import logging
from api.v1.modules.a76.tenants.service import TenantService
from api.v1.modules.a76.user_tenant.service import UserTenantService
from core.config import settings
from fastapi import HTTPException
from keycloak import KeycloakAdmin, KeycloakOpenID
from keycloak.exceptions import KeycloakError
from sqlalchemy.orm import Session
from .dto import (
LoginRequestDTO,
LogoutRequestDTO,
RefreshTokenRequestDTO,
RegisterRequestDTO,
RegisterResponseDTO,
TokenResponseDTO,
UserInfoResponseDTO,
)
logger = logging.getLogger(__name__)
class AuthService:
"""Servicio de autenticación"""
def __init__(self, db: Session):
self.db = db
self.keycloak_openid = KeycloakOpenID(
server_url=settings.KEYCLOAK_SERVER_URL,
client_id=settings.KEYCLOAK_CLIENT_ID,
realm_name=settings.KEYCLOAK_REALM,
client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
)
def login(self, login_data: LoginRequestDTO) -> TokenResponseDTO:
"""
Autentica usuario y obtiene tokens
Args:
login_data: Credenciales de login
Returns:
TokenResponseDTO con access_token y refresh_token
Raises:
HTTPException: Si las credenciales son inválidas
"""
try:
# Verificar que el tenant existe
tenant_service = TenantService(self.db)
user_tenant_service = UserTenantService(self.db)
tenant = tenant_service.get_tenant_by_slug(login_data.tenant_slug)
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
if not tenant.is_active:
raise HTTPException(status_code=403, detail="Tenant is not active")
# Crear nueva instancia de KeycloakOpenID con el realm del tenant
keycloak_client = KeycloakOpenID(
server_url=settings.KEYCLOAK_SERVER_URL,
client_id=settings.KEYCLOAK_CLIENT_ID,
realm_name=tenant.keycloak_realm,
client_secret_key=settings.KEYCLOAK_CLIENT_SECRET,
)
# PASO 1: Primero actualizamos los atributos del usuario ANTES de autenticar
# Esto es necesario para que los Protocol Mappers incluyan los valores correctos
# en el token que se generará a continuación
# Para obtener el user_id, necesitamos hacer una autenticación temporal
# o buscar el usuario por username
try:
keycloak_admin = KeycloakAdmin(
server_url=settings.KEYCLOAK_SERVER_URL,
username=settings.KEYCLOAK_ADMIN_USERNAME,
password=settings.KEYCLOAK_ADMIN_PASSWORD,
realm_name=tenant.keycloak_realm,
user_realm_name="master",
verify=True,
)
# Buscar usuario por username
users = keycloak_admin.get_users({"username": login_data.username})
if users and len(users) > 0:
user_id = users[0]["id"]
# Verificar si el usuario tiene acceso a este tenant
has_access = user_tenant_service.user_has_access_to_tenant(
user_id, tenant.id
)
if not has_access:
logger.warning(
f"User {user_id} tried to access tenant {tenant.id} without permission"
)
raise HTTPException(
status_code=403,
detail="You don't have access to this tenant",
)
# Obtener los datos actuales del usuario
current_user = keycloak_admin.get_user(user_id)
current_attributes = current_user.get("attributes", {})
# Actualizar los atributos de tenant
current_attributes["tenant_id"] = [str(tenant.id)]
current_attributes["tenant_slug"] = [tenant.slug]
# Actualizar el usuario con los nuevos atributos
update_payload = {
"email": current_user.get("email"),
"firstName": current_user.get("firstName"),
"lastName": current_user.get("lastName"),
"enabled": current_user.get("enabled", True),
"emailVerified": current_user.get("emailVerified", False),
"attributes": current_attributes,
}
keycloak_admin.update_user(user_id=user_id, payload=update_payload)
except KeycloakError as e:
logger.warning(f"Could not pre-update user attributes: {str(e)}")
# Continuamos con el login aunque falle la actualización
except HTTPException:
raise # Re-lanzamos las excepciones HTTP (como acceso denegado)
except Exception as e:
logger.warning(f"Error pre-updating user attributes: {str(e)}")
# PASO 2: Ahora autenticamos al usuario
# Si los Protocol Mappers están configurados, el token incluirá
# automáticamente los atributos tenant_id y tenant_slug actualizados
token_response = keycloak_client.token(
username=login_data.username,
password=login_data.password,
grant_type=["password"],
)
return TokenResponseDTO(
access_token=token_response["access_token"],
refresh_token=token_response["refresh_token"],
token_type="bearer",
expires_in=token_response["expires_in"],
)
except KeycloakError as e:
logger.warning(f"Keycloak authentication failed: {str(e)}")
raise HTTPException(status_code=401, detail="Invalid credentials")
except HTTPException:
raise
except Exception as e:
logger.error(f"Login error: {str(e)}")
raise HTTPException(status_code=500, detail="Authentication error")
def refresh_token(self, refresh_data: RefreshTokenRequestDTO) -> TokenResponseDTO:
"""
Refresca el access token usando refresh token
Args:
refresh_data: Refresh token
Returns:
TokenResponseDTO con nuevos tokens
"""
try:
token_response = self.keycloak_openid.refresh_token(
refresh_data.refresh_token
)
return TokenResponseDTO(
access_token=token_response["access_token"],
refresh_token=token_response["refresh_token"],
token_type="bearer",
expires_in=token_response["expires_in"],
)
except KeycloakError as e:
logger.warning(f"Token refresh failed: {str(e)}")
raise HTTPException(
status_code=401, detail="Invalid or expired refresh token"
)
except Exception as e:
logger.error(f"Token refresh error: {str(e)}")
raise HTTPException(status_code=500, detail="Token refresh error")
def get_user_info(self, access_token: str) -> UserInfoResponseDTO:
"""
Obtiene información del usuario desde el token
Args:
access_token: Access token JWT
Returns:
UserInfoResponseDTO con información del usuario
"""
try:
user_info = self.keycloak_openid.userinfo(access_token)
# Extraer roles
roles = []
if "realm_access" in user_info:
roles = user_info["realm_access"].get("roles", [])
# Extraer tenant_id si está presente
tenant_id = user_info.get("tenant_id")
if not tenant_id and "attributes" in user_info:
tenant_id = user_info["attributes"].get("tenant_id")
return UserInfoResponseDTO(
sub=user_info.get("sub"),
email=user_info.get("email"),
name=user_info.get("name"),
preferred_username=user_info.get("preferred_username"),
tenant_id=int(tenant_id) if tenant_id else None,
roles=roles,
)
except KeycloakError as e:
logger.warning(f"Get user info failed: {str(e)}")
raise HTTPException(status_code=401, detail="Invalid token")
except Exception as e:
logger.error(f"Get user info error: {str(e)}")
raise HTTPException(status_code=500, detail="Error retrieving user info")
def logout(self, logout_data: LogoutRequestDTO) -> dict:
"""
Cierra sesión invalidando el refresh token
Args:
logout_data: Refresh token a invalidar
Returns:
Dict con mensaje de éxito
"""
try:
self.keycloak_openid.logout(logout_data.refresh_token)
return {"message": "Logged out successfully"}
except KeycloakError as e:
logger.warning(f"Logout failed: {str(e)}")
# No lanzamos error aquí, el logout puede fallar si el token ya expiró
return {"message": "Logged out"}
except Exception as e:
logger.error(f"Logout error: {str(e)}")
raise HTTPException(status_code=500, detail="Logout error")
def register(self, register_data: RegisterRequestDTO) -> RegisterResponseDTO:
"""
Registra un nuevo usuario en Keycloak
Args:
register_data: Datos del usuario a registrar
Returns:
RegisterResponseDTO con información del usuario creado
Raises:
HTTPException: Si el registro falla
"""
try:
# Verificar que el tenant existe
from api.v1.modules.a76.tenants.service import TenantService
tenant_service = TenantService(self.db)
tenant = tenant_service.get_tenant_by_slug(register_data.tenant_slug)
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
if not tenant.is_active:
raise HTTPException(status_code=403, detail="Tenant is not active")
# Crear instancia de KeycloakAdmin para gestión de usuarios
keycloak_admin = KeycloakAdmin(
server_url=settings.KEYCLOAK_SERVER_URL,
username=settings.KEYCLOAK_ADMIN_USERNAME,
password=settings.KEYCLOAK_ADMIN_PASSWORD,
realm_name=tenant.keycloak_realm,
user_realm_name="master", # El admin suele estar en master realm
verify=True,
)
# Preparar datos del usuario para Keycloak
user_data = {
"username": register_data.username,
"email": register_data.email,
"firstName": register_data.first_name,
"lastName": register_data.last_name,
"enabled": True,
"emailVerified": False,
"credentials": [
{
"type": "password",
"value": register_data.password,
"temporary": False,
}
],
"attributes": {"tenant_id": str(tenant.id), "tenant_slug": tenant.slug},
}
# Crear usuario en Keycloak
user_id = keycloak_admin.create_user(user_data)
# Asignar rol por defecto (user) - opcional, solo si existe
try:
user_role = keycloak_admin.get_realm_role("user")
if user_role:
keycloak_admin.assign_realm_roles(user_id, [user_role])
except KeycloakError as e:
# El rol 'user' no existe, no es un error crítico
logger.warning(f"Could not assign 'user' role: {str(e)}")
# Agregar el usuario al tenant en la base de datos
try:
from api.v1.modules.a76.user_tenant.service import UserTenantService
user_tenant_service = UserTenantService(self.db)
user_tenant_service.add_user_to_tenant(
keycloak_user_id=user_id,
tenant_id=tenant.id,
role="user", # Rol por defecto
)
except Exception as e:
# Si falla, hacer rollback del usuario en Keycloak
logger.error(f"Failed to add user to tenant in database: {str(e)}")
try:
keycloak_admin.delete_user(user_id)
except Exception as e:
pass
raise HTTPException(
status_code=500, detail="Failed to register user in database"
)
return RegisterResponseDTO(
user_id=user_id,
username=register_data.username,
email=register_data.email,
message="User registered successfully",
)
except KeycloakError as e:
error_message = str(e)
logger.warning(f"Keycloak registration failed: {error_message}")
# Mensajes de error más específicos
if "User exists" in error_message or "409" in error_message:
raise HTTPException(
status_code=409, detail="Username or email already exists"
)
elif "Invalid" in error_message:
raise HTTPException(status_code=400, detail="Invalid user data")
else:
raise HTTPException(status_code=500, detail="Registration error")
except HTTPException:
raise
except Exception as e:
logger.error(f"Registration error: {str(e)}")
raise HTTPException(status_code=500, detail="Registration error")
def exchange_code(self, exchange_data) -> TokenResponseDTO:
"""
Intercambia un authorization code por tokens
Este método se usa cuando el frontend recibe un código de autorización
después de un login con proveedor externo (Microsoft, Google, etc.)
a través de Keycloak.
Args:
exchange_data: Datos del código y redirect_uri
Returns:
TokenResponseDTO con access_token y refresh_token
Raises:
HTTPException: Si el código es inválido o expiró
"""
try:
# Importar el DTO aquí para evitar referencias circulares
# Intercambiar código por tokens usando Keycloak
token_response = self.keycloak_openid.token(
grant_type="authorization_code",
code=exchange_data.code,
redirect_uri=exchange_data.redirect_uri,
)
# Si se proporciona tenant_slug, podríamos validar que el usuario pertenece a ese tenant
# Por ahora simplemente retornamos los tokens
if exchange_data.tenant_slug:
# Validar que el tenant existe y está activo
tenant_service = TenantService(self.db)
tenant = tenant_service.get_tenant_by_slug(exchange_data.tenant_slug)
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
if not tenant.is_active:
raise HTTPException(status_code=403, detail="Tenant is not active")
# Opcional: Verificar que el usuario pertenece al tenant
# Esto depende de cómo manejes los tenants en tu aplicación
return TokenResponseDTO(
access_token=token_response["access_token"],
refresh_token=token_response["refresh_token"],
token_type=token_response.get("token_type", "bearer"),
expires_in=token_response.get("expires_in", 3600),
)
except KeycloakError as e:
error_message = str(e)
logger.warning(f"Code exchange failed: {error_message}")
if "invalid_grant" in error_message.lower():
raise HTTPException(
status_code=400, detail="Invalid or expired authorization code"
)
elif "invalid_client" in error_message.lower():
raise HTTPException(
status_code=401, detail="Invalid client credentials"
)
else:
raise HTTPException(status_code=500, detail="Token exchange error")
except HTTPException:
raise
except Exception as e:
logger.error(f"Code exchange error: {str(e)}")
raise HTTPException(status_code=500, detail="Code exchange error")

2
backend/api/v1/modules/a76/general_catalogs/company/models.py
View File

@@ -31,7 +31,7 @@ class Company(Base, TimestampMixin):
# Primary key
id: Mapped[int] = mapped_column(Integer, primary_key=True)
tenant_id: Mapped[int] = mapped_column(Integer, ForeignKey("a76.tenants.id"), nullable=False, index=True)
tenant_id: Mapped[int] = mapped_column(Integer, ForeignKey("core.tenants.id"), nullable=False, index=True)
# Información básica de la empresa
name: Mapped[Optional[str]] = mapped_column(String(255))

7
backend/api/v1/modules/a76/licenses/__init__.py
View File

@@ -1,7 +0,0 @@
"""
Módulo de Licenses
"""
from .routes import router
__all__ = ["router"]

156
backend/api/v1/modules/a76/licenses/dto.py
View File

@@ -1,156 +0,0 @@
"""
DTOs para módulo de licencias
"""
from datetime import datetime
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field
class LicensePlanDTO(str, Enum):
"""Planes de licencia"""
FREE = "free"
BASIC = "basic"
PROFESSIONAL = "professional"
ENTERPRISE = "enterprise"
class LicenseStatusDTO(str, Enum):
"""Estados de licencia"""
ACTIVE = "active"
EXPIRED = "expired"
SUSPENDED = "suspended"
PENDING = "pending"
CANCELLED = "cancelled"
class LicenseCreateDTO(BaseModel):
"""DTO para crear una nueva licencia"""
tenant_id: int = Field(..., description="ID del tenant")
plan: LicensePlanDTO = Field(..., description="Plan de licencia")
max_users: int = Field(default=5, ge=1, description="Número máximo de usuarios")
max_storage_gb: int = Field(
default=10, ge=1, description="Almacenamiento máximo en GB"
)
max_monthly_operations: int = Field(
default=1000, ge=1, description="Operaciones mensuales máximas"
)
feature_api_access: bool = Field(default=True)
feature_advanced_reports: bool = Field(default=False)
feature_integrations: bool = Field(default=False)
feature_dedicated_support: bool = Field(default=False)
starts_at: datetime = Field(..., description="Fecha de inicio de vigencia")
expires_at: datetime = Field(..., description="Fecha de expiración")
class Config:
json_schema_extra = {
"example": {
"tenant_id": 1,
"plan": "professional",
"max_users": 20,
"max_storage_gb": 100,
"max_monthly_operations": 10000,
"feature_api_access": True,
"feature_advanced_reports": True,
"feature_integrations": True,
"feature_dedicated_support": False,
"starts_at": "2025-01-01T00:00:00Z",
"expires_at": "2025-12-31T23:59:59Z",
}
}
class LicenseUpdateDTO(BaseModel):
"""DTO para actualizar una licencia"""
plan: Optional[LicensePlanDTO] = None
status: Optional[LicenseStatusDTO] = None
max_users: Optional[int] = Field(None, ge=1)
max_storage_gb: Optional[int] = Field(None, ge=1)
max_monthly_operations: Optional[int] = Field(None, ge=1)
feature_api_access: Optional[bool] = None
feature_advanced_reports: Optional[bool] = None
feature_integrations: Optional[bool] = None
feature_dedicated_support: Optional[bool] = None
expires_at: Optional[datetime] = None
class LicenseResponseDTO(BaseModel):
"""DTO para respuesta de licencia"""
id: int
tenant_id: int
plan: LicensePlanDTO
status: LicenseStatusDTO
max_users: int
max_storage_gb: int
max_monthly_operations: int
feature_api_access: bool
feature_advanced_reports: bool
feature_integrations: bool
feature_dedicated_support: bool
starts_at: datetime
expires_at: datetime
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class LicenseValidationResponseDTO(BaseModel):
"""DTO para respuesta de validación de licencia"""
is_valid: bool
status: LicenseStatusDTO
plan: LicensePlanDTO
expires_at: datetime
reason: Optional[str] = None
class Config:
json_schema_extra = {
"example": {
"is_valid": True,
"status": "active",
"plan": "professional",
"expires_at": "2025-12-31T23:59:59Z",
"reason": None,
}
}
class LicenseUsageResponseDTO(BaseModel):
"""DTO para respuesta de uso de licencia"""
tenant_id: int
period_start: datetime
period_end: datetime
active_users: int
storage_used_gb: int
operations_count: int
api_calls_count: int
# Límites actuales
max_users: int
max_storage_gb: int
max_monthly_operations: int
# Porcentajes de uso
users_usage_percent: float
storage_usage_percent: float
operations_usage_percent: float
class Config:
from_attributes = True

94
backend/api/v1/modules/a76/licenses/models.py
View File

@@ -1,94 +0,0 @@
"""
Modelos ORM para gestión de licencias
"""
import enum
from api.v1.common.base_models import TimestampMixin
from core.database import Base
from sqlalchemy import Boolean, Column, DateTime
from sqlalchemy import Enum as SQLEnum
from sqlalchemy import ForeignKey, Integer
class LicensePlan(enum.Enum):
"""Planes de licencia disponibles"""
FREE = "free"
BASIC = "basic"
PROFESSIONAL = "professional"
ENTERPRISE = "enterprise"
class LicenseStatus(enum.Enum):
"""Estados de licencia"""
ACTIVE = "active"
EXPIRED = "expired"
SUSPENDED = "suspended"
PENDING = "pending"
CANCELLED = "cancelled"
class License(Base, TimestampMixin):
"""
Modelo de Licencia - Control de planes y límites por tenant
"""
__tablename__ = "licenses"
__table_args__ = {"schema": "a76"}
id = Column(Integer, primary_key=True, index=True)
tenant_id = Column(
Integer, ForeignKey("a76.tenants.id"), nullable=False, unique=True, index=True
)
# Plan y características
plan = Column(SQLEnum(LicensePlan), default=LicensePlan.FREE, nullable=False)
status = Column(
SQLEnum(LicenseStatus), default=LicenseStatus.PENDING, nullable=False
)
# Límites del plan
max_users = Column(Integer, default=5, nullable=False)
max_storage_gb = Column(Integer, default=10, nullable=False)
max_monthly_operations = Column(Integer, default=1000, nullable=False)
# Features habilitadas (booleans)
feature_api_access = Column(Boolean, default=True)
feature_advanced_reports = Column(Boolean, default=False)
feature_integrations = Column(Boolean, default=False)
feature_dedicated_support = Column(Boolean, default=False)
# Vigencia
starts_at = Column(DateTime(timezone=True), nullable=False)
expires_at = Column(DateTime(timezone=True), nullable=False)
def __repr__(self):
return f"<License(tenant_id={self.tenant_id}, plan={self.plan.value}, status={self.status.value})>"
class LicenseUsage(Base, TimestampMixin):
"""
Modelo para tracking de uso de licencia
"""
__tablename__ = "license_usage"
__table_args__ = {"schema": "a76"}
id = Column(Integer, primary_key=True, index=True)
tenant_id = Column(
Integer, ForeignKey("a76.tenants.id"), nullable=False, index=True
)
# Métricas de uso
period_start = Column(DateTime(timezone=True), nullable=False)
period_end = Column(DateTime(timezone=True), nullable=False)
active_users = Column(Integer, default=0)
storage_used_gb = Column(Integer, default=0)
operations_count = Column(Integer, default=0)
api_calls_count = Column(Integer, default=0)
def __repr__(self):
return f"<LicenseUsage(tenant_id={self.tenant_id}, operations={self.operations_count})>"

119
backend/api/v1/modules/a76/licenses/routes.py
View File

@@ -1,119 +0,0 @@
"""
Endpoints API para gestión de licencias
"""
from core.database import get_core_db
from core.security import get_current_user, has_role
from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy.orm import Session
from .dto import (
LicenseCreateDTO,
LicenseResponseDTO,
LicenseUpdateDTO,
LicenseUsageResponseDTO,
LicenseValidationResponseDTO,
)
from .service import LicenseService
router = APIRouter(prefix="/licenses")
@router.post("/", response_model=LicenseResponseDTO, status_code=201)
async def create_license(
license_data: LicenseCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin")),
):
"""
Crea una nueva licencia para un tenant
Requiere rol: admin
"""
service = LicenseService(db)
return service.create_license(license_data)
@router.get("/tenant/{tenant_id}", response_model=LicenseResponseDTO)
async def get_license_by_tenant(
tenant_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Obtiene la licencia de un tenant específico
"""
service = LicenseService(db)
license = service.get_license_by_tenant(tenant_id)
if not license:
raise HTTPException(status_code=404, detail="License not found")
return license
@router.put("/tenant/{tenant_id}", response_model=LicenseResponseDTO)
async def update_license(
tenant_id: int,
license_data: LicenseUpdateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin")),
):
"""
Actualiza la licencia de un tenant
Requiere rol: admin
"""
service = LicenseService(db)
license = service.update_license(tenant_id, license_data)
if not license:
raise HTTPException(status_code=404, detail="License not found")
return license
@router.get("/validate/{tenant_id}", response_model=LicenseValidationResponseDTO)
async def validate_license(
tenant_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Valida si la licencia de un tenant está activa y vigente
"""
service = LicenseService(db)
validation = service.validate_license(tenant_id)
return LicenseValidationResponseDTO(**validation)
@router.get("/usage/{tenant_id}", response_model=LicenseUsageResponseDTO)
async def get_license_usage(
tenant_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Obtiene el uso actual de la licencia de un tenant
"""
service = LicenseService(db)
usage = service.get_usage(tenant_id)
if not usage:
raise HTTPException(status_code=404, detail="License not found")
return usage
@router.get("/my-license", response_model=LicenseResponseDTO)
async def get_my_license(
request: Request,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Obtiene la licencia del tenant del usuario actual
"""
tenant_id = getattr(request.state, "tenant_id", None)
if not tenant_id:
raise HTTPException(status_code=400, detail="Tenant ID not found in request")
service = LicenseService(db)
license = service.get_license_by_tenant(tenant_id)
if not license:
raise HTTPException(status_code=404, detail="License not found")
return license

261
backend/api/v1/modules/a76/licenses/service.py
View File

@@ -1,261 +0,0 @@
"""
Servicio de lógica de negocio para licencias
"""
import logging
from datetime import datetime, timezone
from typing import Optional
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from .dto import (
LicenseCreateDTO,
LicenseResponseDTO,
LicenseUpdateDTO,
LicenseUsageResponseDTO,
)
from .models import License, LicensePlan, LicenseStatus, LicenseUsage
logger = logging.getLogger(__name__)
class LicenseService:
"""Servicio para gestión de licencias"""
def __init__(self, db: Session):
self.db = db
def create_license(self, license_data: LicenseCreateDTO) -> LicenseResponseDTO:
"""
Crea una nueva licencia para un tenant
Args:
license_data: Datos de la licencia
Returns:
LicenseResponseDTO
Raises:
HTTPException: Si el tenant ya tiene licencia o hay error
"""
try:
# Verificar que el tenant no tenga ya una licencia
existing = (
self.db.query(License)
.filter(License.tenant_id == license_data.tenant_id)
.first()
)
if existing:
raise HTTPException(
status_code=400,
detail=f"Tenant {license_data.tenant_id} already has a license",
)
# Crear licencia
db_license = License(
tenant_id=license_data.tenant_id,
plan=LicensePlan(license_data.plan.value),
status=LicenseStatus.ACTIVE,
max_users=license_data.max_users,
max_storage_gb=license_data.max_storage_gb,
max_monthly_operations=license_data.max_monthly_operations,
feature_api_access=license_data.feature_api_access,
feature_advanced_reports=license_data.feature_advanced_reports,
feature_integrations=license_data.feature_integrations,
feature_dedicated_support=license_data.feature_dedicated_support,
starts_at=license_data.starts_at,
expires_at=license_data.expires_at,
)
self.db.add(db_license)
self.db.commit()
self.db.refresh(db_license)
return LicenseResponseDTO.model_validate(db_license)
except IntegrityError as e:
self.db.rollback()
logger.error(f"IntegrityError creating license: {str(e)}")
raise HTTPException(status_code=400, detail="Database integrity error")
except HTTPException:
raise
except Exception as e:
self.db.rollback()
logger.error(f"Error creating license: {str(e)}")
raise HTTPException(status_code=500, detail="Error creating license")
def get_license_by_tenant(self, tenant_id: int) -> Optional[LicenseResponseDTO]:
"""
Obtiene la licencia de un tenant
Args:
tenant_id: ID del tenant
Returns:
LicenseResponseDTO o None si no existe
"""
license = self.db.query(License).filter(License.tenant_id == tenant_id).first()
if not license:
return None
return LicenseResponseDTO.model_validate(license)
def update_license(
self, tenant_id: int, license_data: LicenseUpdateDTO
) -> Optional[LicenseResponseDTO]:
"""
Actualiza una licencia
Args:
tenant_id: ID del tenant
license_data: Datos a actualizar
Returns:
LicenseResponseDTO actualizado o None si no existe
"""
license = self.db.query(License).filter(License.tenant_id == tenant_id).first()
if not license:
return None
# Actualizar campos proporcionados
update_data = license_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
if field in ["plan", "status"]:
# Convertir enums
value = LicensePlan(value) if field == "plan" else LicenseStatus(value)
setattr(license, field, value)
try:
self.db.commit()
self.db.refresh(license)
logger.info(f"License updated for tenant {tenant_id}")
return LicenseResponseDTO.model_validate(license)
except Exception as e:
self.db.rollback()
logger.error(f"Error updating license for tenant {tenant_id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error updating license")
def validate_license(self, tenant_id: int) -> dict:
"""
Valida si la licencia de un tenant está activa y vigente
Args:
tenant_id: ID del tenant
Returns:
Dict con información de validación
"""
license = self.db.query(License).filter(License.tenant_id == tenant_id).first()
if not license:
return {
"is_valid": False,
"status": "not_found",
"plan": None,
"expires_at": None,
"reason": "License not found",
}
now = datetime.now(timezone.utc)
# Verificar estado
if license.status != LicenseStatus.ACTIVE:
return {
"is_valid": False,
"status": license.status.value,
"plan": license.plan.value,
"expires_at": license.expires_at,
"reason": f"License status is {license.status.value}",
}
# Verificar vigencia
if license.expires_at < now:
# Auto-actualizar a expirada
license.status = LicenseStatus.EXPIRED
self.db.commit()
return {
"is_valid": False,
"status": "expired",
"plan": license.plan.value,
"expires_at": license.expires_at,
"reason": "License has expired",
}
# Licencia válida
return {
"is_valid": True,
"status": license.status.value,
"plan": license.plan.value,
"expires_at": license.expires_at,
"reason": None,
}
def get_usage(self, tenant_id: int) -> Optional[LicenseUsageResponseDTO]:
"""
Obtiene el uso actual de la licencia de un tenant
Args:
tenant_id: ID del tenant
Returns:
LicenseUsageResponseDTO o None
"""
license = self.db.query(License).filter(License.tenant_id == tenant_id).first()
if not license:
return None
# Obtener último registro de uso
usage = (
self.db.query(LicenseUsage)
.filter(LicenseUsage.tenant_id == tenant_id)
.order_by(LicenseUsage.created_at.desc())
.first()
)
if not usage:
# Crear registro inicial si no existe
usage = LicenseUsage(
tenant_id=tenant_id,
period_start=datetime.now(timezone.utc),
period_end=datetime.now(timezone.utc),
active_users=0,
storage_used_gb=0,
operations_count=0,
api_calls_count=0,
)
# Calcular porcentajes
users_usage = (
(usage.active_users / license.max_users * 100)
if license.max_users > 0
else 0
)
storage_usage = (
(usage.storage_used_gb / license.max_storage_gb * 100)
if license.max_storage_gb > 0
else 0
)
operations_usage = (
(usage.operations_count / license.max_monthly_operations * 100)
if license.max_monthly_operations > 0
else 0
)
return LicenseUsageResponseDTO(
tenant_id=tenant_id,
period_start=usage.period_start,
period_end=usage.period_end,
active_users=usage.active_users,
storage_used_gb=usage.storage_used_gb,
operations_count=usage.operations_count,
api_calls_count=usage.api_calls_count,
max_users=license.max_users,
max_storage_gb=license.max_storage_gb,
max_monthly_operations=license.max_monthly_operations,
users_usage_percent=round(users_usage, 2),
storage_usage_percent=round(storage_usage, 2),
operations_usage_percent=round(operations_usage, 2),
)

8
backend/api/v1/modules/a76/router.py
View File

@@ -8,7 +8,7 @@ from fastapi import APIRouter
from .customs_brokers.routes import router as customs_broker_router
# Importar routers de módulos
from .auth import router as auth_router
from ..core.auth import router as auth_router
from .classes import router as classes_router
from .clients_and_providers import router as client_and_provider_router
from .general_catalogs.company import router as company_router
@@ -17,7 +17,7 @@ from .transportation.drivers.routes import router as drivers_router
from .general_catalogs.exchange_rate.routes import router as exchange_rate_router
from .general_catalogs.identifiers.routes import router as identifiers_router
from .fraction_rule_octave.routes import router as fraction_rule_octave_router
from .licenses import router as licenses_router
from ..core.licenses import router as licenses_router
from .general_catalogs.packages.routes import router as package_router
from .general_catalogs.ports.routes import router as ports_router
from .parts import router as parts_router
@@ -38,10 +38,10 @@ from .general_catalogs.error_catalogs.routes import router as error_catalogs_rou
from .general_catalogs.doda.routes import router as doda_router
from .general_catalogs.prevalidators.routes import router as prevalidators_router
from .general_catalogs.electronic_notices.routes import router as electronic_notices_router
from .tenants import router as tenants_router
from ..core.tenants import router as tenants_router
from .transportation.trailers.routes import router as trailers_router
from .transportation.transporters.routes import router as transporters_router
from .user_tenant.routes import router as user_tenant_router
from ..core.user_tenant.routes import router as user_tenant_router
from .transportation.vehicles.routes import router as vehicles_router
# Router principal

7
backend/api/v1/modules/a76/tenants/__init__.py
View File

@@ -1,7 +0,0 @@
"""
Módulo de Tenants
"""
from .routes import router
__all__ = ["router"]

116
backend/api/v1/modules/a76/tenants/dto.py
View File

@@ -1,116 +0,0 @@
"""
DTOs (Data Transfer Objects) para módulo de tenants
Reemplaza schemas.py siguiendo enfoque DDD y estilo NestJS
"""
from datetime import datetime
from enum import Enum
from typing import Optional
from pydantic import BaseModel, EmailStr, Field
class TenantTypeDTO(str, Enum):
"""Tipo de tenant"""
SHARED = "shared"
DEDICATED = "dedicated"
class TenantCreateDTO(BaseModel):
"""DTO para crear un nuevo tenant"""
name: str = Field(
..., min_length=3, max_length=255, description="Nombre del tenant"
)
slug: str = Field(
..., min_length=3, max_length=100, description="Identificador único del tenant"
)
keycloak_realm: str = Field(
..., min_length=3, max_length=255, description="Nombre del realm en Keycloak"
)
type: TenantTypeDTO = Field(
default=TenantTypeDTO.SHARED, description="Tipo de tenant"
)
contact_name: Optional[str] = Field(
None, max_length=255, description="Nombre de contacto"
)
contact_email: Optional[EmailStr] = Field(None, description="Email de contacto")
contact_phone: Optional[str] = Field(
None, max_length=50, description="Teléfono de contacto"
)
class Config:
json_schema_extra = {
"example": {
"name": "Empresa ABC S.A. de C.V.",
"slug": "empresa-abc",
"keycloak_realm": "empresa-abc-realm",
"type": "shared",
"contact_name": "Juan Pérez",
"contact_email": "juan.perez@empresa-abc.com",
"contact_phone": "+52 55 1234 5678",
}
}
class TenantUpdateDTO(BaseModel):
"""DTO para actualizar un tenant"""
name: Optional[str] = Field(None, min_length=3, max_length=255)
contact_name: Optional[str] = Field(None, max_length=255)
contact_email: Optional[EmailStr] = None
contact_phone: Optional[str] = Field(None, max_length=50)
is_active: Optional[bool] = None
class Config:
json_schema_extra = {
"example": {
"name": "Empresa ABC S.A. de C.V. - Actualizado",
"contact_email": "nuevo@empresa-abc.com",
}
}
class TenantResponseDTO(BaseModel):
"""DTO para respuesta de tenant"""
id: int
name: str
slug: str
type: TenantTypeDTO
keycloak_realm: str
contact_name: Optional[str]
contact_email: Optional[str]
contact_phone: Optional[str]
is_active: bool
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
json_schema_extra = {
"example": {
"id": 1,
"name": "Empresa ABC S.A. de C.V.",
"slug": "empresa-abc",
"type": "shared",
"keycloak_realm": "empresa-abc-realm",
"contact_name": "Juan Pérez",
"contact_email": "juan.perez@empresa-abc.com",
"contact_phone": "+52 55 1234 5678",
"is_active": True,
"created_at": "2025-01-15T10:30:00Z",
"updated_at": "2025-01-15T10:30:00Z",
}
}
class TenantListResponseDTO(BaseModel):
"""DTO para lista de tenants"""
tenants: list[TenantResponseDTO]
total: int
page: int
page_size: int

62
backend/api/v1/modules/a76/tenants/models.py
View File

@@ -1,62 +0,0 @@
"""
Modelos ORM para gestión de tenants
"""
import enum
from typing import TYPE_CHECKING, List
from api.v1.common.base_models import TimestampMixin
from core.database import Base
from sqlalchemy import Boolean, Column
from sqlalchemy import Enum as SQLEnum
from sqlalchemy import Integer, String, Text
from sqlalchemy.orm import Mapped, relationship
if TYPE_CHECKING:
from api.v1.modules.a76.user_tenant.models import UserTenant
class TenantType(enum.Enum):
"""Tipo de tenant según tamaño y necesidades"""
SHARED = "shared" # BD compartida
DEDICATED = "dedicated" # BD dedicada
class Tenant(Base, TimestampMixin):
"""
Modelo de Tenant - Cliente/Organización en el sistema
Cada tenant puede tener BD compartida o dedicada
"""
__tablename__ = "tenants"
__table_args__ = {"schema": "a76", "extend_existing": True}
id = Column(Integer, primary_key=True, index=True)
name = Column(String(255), nullable=False, index=True)
slug = Column(String(100), unique=True, nullable=False, index=True)
# Tipo de tenant (compartido o dedicado)
type = Column(SQLEnum(TenantType), default=TenantType.SHARED, nullable=False)
# Keycloak realm asociado
keycloak_realm = Column(String(255), nullable=False)
# Configuración de BD dedicada (JSON string o NULL si usa BD compartida)
db_config = Column(Text, nullable=True) # JSON: {host, port, name, user, password}
# Información de contacto
contact_name = Column(String(255))
contact_email = Column(String(255))
contact_phone = Column(String(50))
# Estado
is_active = Column(Boolean, default=True, nullable=False)
# Relación con UserTenant
user_relations: Mapped[List["UserTenant"]] = relationship(
"UserTenant", back_populates="tenant"
)
def __repr__(self):
return f"<Tenant(id={self.id}, name={self.name}, type={self.type.value})>"

131
backend/api/v1/modules/a76/tenants/routes.py
View File

@@ -1,131 +0,0 @@
"""
Endpoints API para gestión de tenants
"""
from core.database import get_core_db
from core.security import get_current_user, has_role
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from .dto import (
TenantCreateDTO,
TenantListResponseDTO,
TenantResponseDTO,
TenantUpdateDTO,
)
from .service import TenantService
router = APIRouter(prefix="/tenants")
@router.post("/", response_model=TenantResponseDTO, status_code=201)
async def create_tenant(
tenant_data: TenantCreateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin")),
):
"""
Crea un nuevo tenant en el sistema
Requiere rol: admin
"""
service = TenantService(db)
return service.create_tenant(tenant_data)
@router.get("/", response_model=TenantListResponseDTO)
async def list_tenants(
page: int = Query(1, ge=1, description="Número de página"),
page_size: int = Query(50, ge=1, le=100, description="Tamaño de página"),
active_only: bool = Query(False, description="Solo tenants activos"),
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin")),
):
"""
Lista todos los tenants
Requiere rol: admin
"""
service = TenantService(db)
skip = (page - 1) * page_size
tenants = service.list_tenants(skip=skip, limit=page_size, active_only=active_only)
# Contar total
from .models import Tenant
query = db.query(Tenant)
if active_only:
query = query.filter(Tenant.is_active)
total = query.count()
return TenantListResponseDTO(
tenants=tenants, total=total, page=page, page_size=page_size
)
@router.get("/{tenant_id}", response_model=TenantResponseDTO)
async def get_tenant(
tenant_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Obtiene información de un tenant por ID
"""
service = TenantService(db)
tenant = service.get_tenant(tenant_id)
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
return tenant
@router.put("/{tenant_id}", response_model=TenantResponseDTO)
async def update_tenant(
tenant_id: int,
tenant_data: TenantUpdateDTO,
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin")),
):
"""
Actualiza un tenant
Requiere rol: admin
"""
service = TenantService(db)
tenant = service.update_tenant(tenant_id, tenant_data)
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
return tenant
@router.delete("/{tenant_id}", status_code=204)
async def delete_tenant(
tenant_id: int,
db: Session = Depends(get_core_db),
current_user: dict = Depends(has_role("admin")),
):
"""
Elimina (desactiva) un tenant
Requiere rol: admin
"""
service = TenantService(db)
if not service.delete_tenant(tenant_id):
raise HTTPException(status_code=404, detail="Tenant not found")
return None
@router.get("/slug/{slug}", response_model=TenantResponseDTO)
async def get_tenant_by_slug(
slug: str,
db: Session = Depends(get_core_db),
current_user: dict = Depends(get_current_user),
):
"""
Obtiene un tenant por su slug
"""
service = TenantService(db)
tenant = service.get_tenant_by_slug(slug)
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
return tenant

212
backend/api/v1/modules/a76/tenants/service.py
View File

@@ -1,212 +0,0 @@
"""
Capa de servicio para lógica de negocio de tenants
"""
import json
import logging
from typing import List, Optional
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from .dto import TenantCreateDTO, TenantResponseDTO, TenantUpdateDTO
from .models import Tenant, TenantType
logger = logging.getLogger(__name__)
class TenantService:
"""Servicio para gestión de tenants"""
def __init__(self, db: Session):
self.db = db
def create_tenant(self, tenant_data: TenantCreateDTO) -> TenantResponseDTO:
"""
Crea un nuevo tenant en el sistema
Args:
tenant_data: Datos del tenant a crear
Returns:
TenantResponseDTO con información del tenant creado
Raises:
HTTPException: Si el slug o realm ya existen
"""
try:
# Verificar que no exista el slug
existing = (
self.db.query(Tenant).filter(Tenant.slug == tenant_data.slug).first()
)
if existing:
raise HTTPException(
status_code=400,
detail=f"Tenant with slug '{tenant_data.slug}' already exists",
)
# Crear tenant
db_tenant = Tenant(
name=tenant_data.name,
slug=tenant_data.slug,
keycloak_realm=tenant_data.keycloak_realm,
type=TenantType(tenant_data.type.value),
contact_name=tenant_data.contact_name,
contact_email=tenant_data.contact_email,
contact_phone=tenant_data.contact_phone,
is_active=True,
)
self.db.add(db_tenant)
self.db.commit()
self.db.refresh(db_tenant)
logger.info(f"Tenant created: {db_tenant.id} - {db_tenant.name}")
return TenantResponseDTO.model_validate(db_tenant)
except IntegrityError as e:
self.db.rollback()
logger.error(f"IntegrityError creating tenant: {str(e)}")
raise HTTPException(
status_code=400, detail="Tenant with this slug or realm already exists"
)
except HTTPException:
raise
except Exception as e:
self.db.rollback()
logger.error(f"Error creating tenant: {str(e)}")
raise HTTPException(status_code=500, detail="Error creating tenant")
def get_tenant(self, tenant_id: int) -> Optional[TenantResponseDTO]:
"""
Obtiene un tenant por ID
Args:
tenant_id: ID del tenant
Returns:
TenantResponseDTO o None si no existe
"""
tenant = self.db.query(Tenant).filter(Tenant.id == tenant_id).first()
if not tenant:
return None
return TenantResponseDTO.model_validate(tenant)
def get_tenant_by_slug(self, slug: str) -> Optional[TenantResponseDTO]:
"""Obtiene un tenant por slug"""
tenant = self.db.query(Tenant).filter(Tenant.slug == slug).first()
if not tenant:
return None
return TenantResponseDTO.model_validate(tenant)
def list_tenants(
self, skip: int = 0, limit: int = 100, active_only: bool = False
) -> List[TenantResponseDTO]:
"""
Lista todos los tenants
Args:
skip: Número de registros a omitir
limit: Número máximo de registros a retornar
active_only: Si True, solo retorna tenants activos
Returns:
Lista de TenantResponseDTO
"""
query = self.db.query(Tenant)
if active_only:
query = query.filter(Tenant.is_active)
tenants = query.offset(skip).limit(limit).all()
return [TenantResponseDTO.model_validate(t) for t in tenants]
def update_tenant(
self, tenant_id: int, tenant_data: TenantUpdateDTO
) -> Optional[TenantResponseDTO]:
"""
Actualiza un tenant
Args:
tenant_id: ID del tenant a actualizar
tenant_data: Datos a actualizar
Returns:
TenantResponseDTO actualizado o None si no existe
"""
tenant = self.db.query(Tenant).filter(Tenant.id == tenant_id).first()
if not tenant:
return None
# Actualizar solo campos proporcionados
update_data = tenant_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(tenant, field, value)
try:
self.db.commit()
self.db.refresh(tenant)
logger.info(f"Tenant updated: {tenant_id}")
return TenantResponseDTO.model_validate(tenant)
except Exception as e:
self.db.rollback()
logger.error(f"Error updating tenant {tenant_id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error updating tenant")
def delete_tenant(self, tenant_id: int) -> bool:
"""
Elimina (desactiva) un tenant
Args:
tenant_id: ID del tenant a eliminar
Returns:
True si se eliminó, False si no existe
"""
tenant = self.db.query(Tenant).filter(Tenant.id == tenant_id).first()
if not tenant:
return False
# Soft delete: marcar como inactivo
tenant.is_active = False
try:
self.db.commit()
logger.info(f"Tenant deleted (soft): {tenant_id}")
return True
except Exception as e:
self.db.rollback()
logger.error(f"Error deleting tenant {tenant_id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error deleting tenant")
def upgrade_to_dedicated(
self, tenant_id: int, db_config: dict
) -> Optional[TenantResponseDTO]:
"""
Actualiza un tenant de BD compartida a BD dedicada
Args:
tenant_id: ID del tenant
db_config: Configuración de BD dedicada
Returns:
TenantResponseDTO actualizado
"""
tenant = self.db.query(Tenant).filter(Tenant.id == tenant_id).first()
if not tenant:
return None
tenant.type = TenantType.DEDICATED
tenant.db_config = json.dumps(db_config)
try:
self.db.commit()
self.db.refresh(tenant)
logger.info(f"Tenant upgraded to dedicated DB: {tenant_id}")
return TenantResponseDTO.model_validate(tenant)
except Exception as e:
self.db.rollback()
logger.error(f"Error upgrading tenant {tenant_id}: {str(e)}")
raise HTTPException(status_code=500, detail="Error upgrading tenant")

67
backend/api/v1/modules/a76/user_tenant/dto.py
View File

@@ -1,67 +0,0 @@
"""
DTOs para gestión de relaciones usuario-tenant
"""
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
class AddUserToTenantRequestDTO(BaseModel):
"""Request para agregar un usuario a un tenant"""
keycloak_user_id: str = Field(..., description="ID del usuario en Keycloak")
tenant_id: int = Field(..., description="ID del tenant")
role: Optional[str] = Field(None, description="Rol del usuario en el tenant")
class RemoveUserFromTenantRequestDTO(BaseModel):
"""Request para eliminar un usuario de un tenant"""
keycloak_user_id: str = Field(..., description="ID del usuario en Keycloak")
tenant_id: int = Field(..., description="ID del tenant")
soft_delete: bool = Field(True, description="Si True, desactiva. Si False, elimina")
class UpdateUserRoleRequestDTO(BaseModel):
"""Request para actualizar el rol de un usuario en un tenant"""
keycloak_user_id: str = Field(..., description="ID del usuario en Keycloak")
tenant_id: int = Field(..., description="ID del tenant")
role: str = Field(..., description="Nuevo rol del usuario")
class UserTenantResponseDTO(BaseModel):
"""Response con información de relación usuario-tenant"""
id: int
keycloak_user_id: str
tenant_id: int
is_active: bool
role: Optional[str]
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True
class TenantBasicInfoDTO(BaseModel):
"""Información básica de un tenant"""
id: int
name: str
slug: str
is_active: bool
keycloak_realm: str
class Config:
from_attributes = True
class UserTenantsResponseDTO(BaseModel):
"""Response con los tenants de un usuario"""
keycloak_user_id: str
tenants: list[TenantBasicInfoDTO]

47
backend/api/v1/modules/a76/user_tenant/models.py
View File

@@ -1,47 +0,0 @@
"""
Modelo de relación entre usuarios (Keycloak) y tenants
"""
from typing import TYPE_CHECKING, Optional
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from sqlalchemy import Boolean, ForeignKeyConstraint, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship
if TYPE_CHECKING:
from api.v1.modules.a76.tenants.models import Tenant
class UserTenant(Base, TenantScopedMixin, TimestampMixin):
"""
Relación muchos-a-muchos entre usuarios de Keycloak y tenants
Un usuario puede pertenecer a múltiples tenants
Un tenant puede tener múltiples usuarios
"""
__tablename__ = "user_tenants"
__table_args__ = (
UniqueConstraint(
"keycloak_user_id", "tenant_id", "company_id", name="uq_user_tenant"
),
{"schema": "a76", "extend_existing": True},
)
# Primary Key
id: Mapped[int] = mapped_column(primary_key=True, index=True)
# ID del usuario en Keycloak (UUID string)
keycloak_user_id: Mapped[str] = mapped_column(
String(255), nullable=False, index=True
)
# Estado de la relación
is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
# Información adicional - Rol del usuario en este tenant (opcional)
role: Mapped[Optional[str]] = mapped_column(String(50), nullable=True)
# Relación con Tenant
tenant: Mapped["Tenant"] = relationship("Tenant", back_populates="user_relations")

141
backend/api/v1/modules/a76/user_tenant/routes.py
View File

@@ -1,141 +0,0 @@
"""
Rutas para gestión de relaciones usuario-tenant
"""
from typing import List
from core.database import get_core_db
from core.security import get_current_user
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from .dto import (
AddUserToTenantRequestDTO,
RemoveUserFromTenantRequestDTO,
TenantBasicInfoDTO,
UpdateUserRoleRequestDTO,
UserTenantResponseDTO,
UserTenantsResponseDTO,
)
from .service import UserTenantService
router = APIRouter(prefix="/user-tenants")
@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,
}

225
backend/api/v1/modules/a76/user_tenant/service.py
View File

@@ -1,225 +0,0 @@
"""
Servicio para gestionar relaciones entre usuarios y tenants
"""
import logging
from typing import List, Optional
from fastapi import HTTPException
from sqlalchemy import and_
from sqlalchemy.orm import Session
from ..tenants.models import Tenant
from .models import UserTenant
logger = logging.getLogger(__name__)
class UserTenantService:
"""Servicio para gestionar acceso de usuarios a tenants"""
def __init__(self, db: Session):
self.db = db
def add_user_to_tenant(
self, keycloak_user_id: str, tenant_id: int, role: Optional[str] = None
) -> UserTenant:
"""
Agrega un usuario a un tenant
Args:
keycloak_user_id: ID del usuario en Keycloak
tenant_id: ID del tenant
role: Rol opcional del usuario en este tenant
Returns:
UserTenant creado
"""
# Verificar que el tenant existe
tenant = self.db.query(Tenant).filter(Tenant.id == tenant_id).first()
if not tenant:
raise HTTPException(status_code=404, detail="Tenant not found")
# Verificar si la relación ya existe
existing = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.tenant_id == tenant_id,
)
)
.first()
)
if existing:
# Si existe pero está inactiva, reactivarla
if not existing.is_active:
existing.is_active = True
existing.role = role
self.db.commit()
self.db.refresh(existing)
return existing
else:
raise HTTPException(
status_code=409, detail="User already has access to this tenant"
)
# Crear nueva relación
user_tenant = UserTenant(
keycloak_user_id=keycloak_user_id,
tenant_id=tenant_id,
role=role,
is_active=True,
)
self.db.add(user_tenant)
self.db.commit()
self.db.refresh(user_tenant)
return user_tenant
def remove_user_from_tenant(
self, keycloak_user_id: str, tenant_id: int, soft_delete: bool = True
) -> bool:
"""
Elimina un usuario de un tenant
Args:
keycloak_user_id: ID del usuario en Keycloak
tenant_id: ID del tenant
soft_delete: Si True, solo marca como inactivo. Si False, elimina físicamente
Returns:
True si se eliminó correctamente
"""
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.tenant_id == tenant_id,
)
)
.first()
)
if not user_tenant:
raise HTTPException(
status_code=404, detail="User-tenant relationship not found"
)
if soft_delete:
user_tenant.is_active = False
self.db.commit()
else:
self.db.delete(user_tenant)
self.db.commit()
return True
def get_user_tenants(self, keycloak_user_id: str) -> List[Tenant]:
"""
Obtiene todos los tenants a los que tiene acceso un usuario
Args:
keycloak_user_id: ID del usuario en Keycloak
Returns:
Lista de tenants
"""
user_tenants = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.is_active,
)
)
.all()
)
tenant_ids = [ut.tenant_id for ut in user_tenants]
tenants = (
self.db.query(Tenant)
.filter(and_(Tenant.id.in_(tenant_ids), Tenant.is_active))
.all()
)
return tenants
def get_tenant_users(self, tenant_id: int) -> List[UserTenant]:
"""
Obtiene todos los usuarios que tienen acceso a un tenant
Args:
tenant_id: ID del tenant
Returns:
Lista de relaciones UserTenant
"""
return (
self.db.query(UserTenant)
.filter(and_(UserTenant.tenant_id == tenant_id, UserTenant.is_active))
.all()
)
def user_has_access_to_tenant(self, keycloak_user_id: str, tenant_id: int) -> bool:
"""
Verifica si un usuario tiene acceso a un tenant
Args:
keycloak_user_id: ID del usuario en Keycloak
tenant_id: ID del tenant
Returns:
True si tiene acceso, False en caso contrario
"""
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.tenant_id == tenant_id,
UserTenant.is_active,
)
)
.first()
)
return user_tenant is not None
def update_user_role_in_tenant(
self, keycloak_user_id: str, tenant_id: int, role: str
) -> UserTenant:
"""
Actualiza el rol de un usuario en un tenant
Args:
keycloak_user_id: ID del usuario en Keycloak
tenant_id: ID del tenant
role: Nuevo rol
Returns:
UserTenant actualizado
"""
user_tenant = (
self.db.query(UserTenant)
.filter(
and_(
UserTenant.keycloak_user_id == keycloak_user_id,
UserTenant.tenant_id == tenant_id,
)
)
.first()
)
if not user_tenant:
raise HTTPException(
status_code=404, detail="User-tenant relationship not found"
)
user_tenant.role = role
self.db.commit()
self.db.refresh(user_tenant)
return user_tenant