acazares
b68c4316ff
- Rearranged imports in multiple files for consistency and clarity. - Updated logging middleware to exclude specific paths from logging. - Enhanced security module by cleaning up token handling and improving tenant validation. - Added tenant and company scoped mixins for better database model management. - Implemented generic CRUD routes for tenant-scoped resources. - Improved error handling and response management in API routes. - Cleaned up login and logout processes to ensure proper session management. - Introduced mechanisms to clear local storage and cookies on tenant change. - Enhanced company store to detect tenant changes and clear data accordingly. - Added new DTO mixins for currency and value affect flags.
262 lines
8.4 KiB
Python
262 lines
8.4 KiB
Python
"""
|
|
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),
|
|
)
|