From fca122759cfb7b244e8cdf029b1e6f22f18da3ac Mon Sep 17 00:00:00 2001 From: acazares Date: Wed, 14 Jan 2026 15:55:44 -0600 Subject: [PATCH] Implement multi-tenant permission management module - Added Pydantic schemas for permission-related requests and responses in `schemas.py`. - Created seed data for various permissions and roles in `seed.py`. - Developed a `PermissionService` class in `service.py` to handle permission logic, including user role assignments and direct permission grants. - Implemented unit tests for permission functionalities in `test_permissions.py`, covering role assignments, permission checks, and expiration handling. - Seeded trailer types data in `trailer_types/seed.py`. --- .../api/v1/modules/core/permissions/README.md | 324 +++++++++++++ .../v1/modules/core/permissions/__init__.py | 38 ++ .../modules/core/permissions/dependencies.py | 231 +++++++++ .../core/permissions/integration_example.py | 153 ++++++ .../api/v1/modules/core/permissions/models.py | 247 ++++++++++ .../api/v1/modules/core/permissions/routes.py | 445 ++++++++++++++++++ .../v1/modules/core/permissions/schemas.py | 245 ++++++++++ .../api/v1/modules/core/permissions/seed.py | 63 +++ .../v1/modules/core/permissions/service.py | 298 ++++++++++++ .../core/permissions/test_permissions.py | 424 +++++++++++++++++ .../reference_data/trailer_types/seed.py | 45 ++ frontend/src/routes/dashboard/+page.svelte | 51 +- .../routes/dashboard/invoices/+page.svelte | 2 +- 13 files changed, 2547 insertions(+), 19 deletions(-) create mode 100644 backend/api/v1/modules/core/permissions/README.md create mode 100644 backend/api/v1/modules/core/permissions/__init__.py create mode 100644 backend/api/v1/modules/core/permissions/dependencies.py create mode 100644 backend/api/v1/modules/core/permissions/integration_example.py create mode 100644 backend/api/v1/modules/core/permissions/models.py create mode 100644 backend/api/v1/modules/core/permissions/routes.py create mode 100644 backend/api/v1/modules/core/permissions/schemas.py create mode 100644 backend/api/v1/modules/core/permissions/seed.py create mode 100644 backend/api/v1/modules/core/permissions/service.py create mode 100644 backend/api/v1/modules/core/permissions/test_permissions.py create mode 100644 backend/api/v1/modules/public/reference_data/trailer_types/seed.py diff --git a/backend/api/v1/modules/core/permissions/README.md b/backend/api/v1/modules/core/permissions/README.md new file mode 100644 index 00000000..d259bca4 --- /dev/null +++ b/backend/api/v1/modules/core/permissions/README.md @@ -0,0 +1,324 @@ +# Módulo de Permisos Multi-Tenant + +Sistema completo de permisos granulares para aplicaciones multi-tenant con FastAPI y SQLAlchemy. + +## 📁 Estructura del Módulo + +``` +backend/api/v1/modules/core/permissions/ +├── __init__.py # Exports del módulo +├── models.py # Modelos SQLAlchemy +├── service.py # Lógica de negocio +├── dependencies.py # Dependencias FastAPI +├── schemas.py # Modelos Pydantic (request/response) +└── routes.py # Endpoints de la API +``` + +## 🎯 Componentes + +### **models.py** + +Define los modelos de base de datos: + +- `Permission` - Permisos del sistema (ej: "invoice.view", "invoice.edit") +- `ClientRole` - Roles personalizados por cliente +- `RolePermission` - Relación roles-permisos +- `UserClientRole` - Asignación usuario-rol-cliente +- `UserClientPermission` - Permisos directos por usuario + +### **service.py** + +Contiene la clase `PermissionService` con métodos: + +- `get_user_permissions()` - Obtiene todos los permisos de un usuario +- `has_permission()` - Verifica un permiso específico +- `has_all_permissions()` - Verifica múltiples permisos (AND) +- `has_any_permission()` - Verifica múltiples permisos (OR) +- `assign_role_to_user()` - Asigna roles a usuarios +- `grant_direct_permission()` - Concede permisos directos + +### **dependencies.py** + +Dependencias para proteger rutas: + +- `PermissionChecker` - Clase para verificar múltiples permisos +- `RequirePermission` - Clase para verificar un solo permiso +- `get_client_id()` - Extrae el ID del cliente del header +- `get_permission_service()` - Proporciona instancia del servicio +- `get_current_user_permissions()` - Devuelve permisos del usuario + +### **schemas.py** + +Modelos Pydantic para request/response: + +- Responses: `PermissionResponse`, `ClientRoleResponse`, `UserPermissionsResponse`, etc. +- Requests: `AssignRoleRequest`, `GrantPermissionRequest`, `CreateRoleRequest`, etc. + +### **routes.py** + +Endpoints de la API: + +- `GET /permissions/me` - Permisos del usuario actual +- `GET /permissions/available` - Lista todos los permisos +- `GET /permissions/roles` - Lista roles del cliente +- `POST /permissions/roles` - Crea un rol +- `POST /permissions/assign-role` - Asigna rol a usuario +- `POST /permissions/grant-permission` - Concede permiso directo +- Ejemplos de rutas protegidas + +## 🚀 Uso Rápido + +### Importar el módulo + +```python +from api.v1.modules.core.permissions import ( + Permission, + ClientRole, + PermissionService, + PermissionChecker, + RequirePermission, + router +) +``` + +### Registrar las rutas + +```python +# En backend/api/v1/router.py +from api.v1.modules.core.permissions import router as permissions_router + +api_router = APIRouter() +api_router.include_router(permissions_router) +``` + +### Proteger una ruta con permiso único + +```python +from fastapi import APIRouter, Depends +from api.v1.modules.core.permissions import RequirePermission + +router = APIRouter() + +@router.get("/invoices") +async def list_invoices( + _: None = Depends(RequirePermission("invoice.view")) +): + return {"invoices": [...]} +``` + +### Proteger con múltiples permisos + +```python +from api.v1.modules.core.permissions import PermissionChecker + +@router.post("/invoices") +async def create_invoice( + _: None = Depends(PermissionChecker( + ["invoice.view", "invoice.create"], + require_all=True # Requiere TODOS + )) +): + return {"created": True} +``` + +### Usar permisos en la lógica + +```python +from api.v1.modules.core.permissions import get_current_user_permissions + +@router.get("/dashboard") +async def dashboard( + permissions: set = Depends(get_current_user_permissions) +): + widgets = [] + + if "invoice.view" in permissions: + widgets.append({"type": "invoices", "data": [...]}) + + return {"widgets": widgets} +``` + +## 📊 Base de Datos + +### Ejecutar migración + +```bash +cd backend +alembic upgrade head +``` + +Esto crea las tablas y permisos iniciales: + +- **invoice.*** - view, create, edit, delete, approve +- **user.*** - view, create, edit, delete +- **report.*** - financial.view, admin.view, export +- **roles.*** - view, create, edit, delete, assign +- **permissions.*** - view, grant + +## 🔐 Flujo de Autenticación + +1. Usuario hace request con token JWT de Keycloak +2. Header `X-Client-ID` indica el cliente/tenant +3. Sistema extrae `user_id` del token +4. Consulta permisos del usuario en ese cliente +5. Valida si tiene el permiso requerido +6. Devuelve 200 OK o 403 Forbidden + +## 💡 Ejemplos Prácticos + +### Crear un rol personalizado + +```python +from api.v1.modules.core.permissions import PermissionService +from core.database import get_db + +db = next(get_db()) +service = PermissionService(db) + +# Crear rol +role = ClientRole( + client_id=1, + name="Contador", + code="accountant", + description="Acceso a módulo contable" +) +db.add(role) +db.commit() +``` + +### Asignar permisos a un rol + +```python +from api.v1.modules.core.permissions.models import RolePermission + +# Obtener permisos de facturación +invoice_perms = db.query(Permission).filter( + Permission.module == "invoice" +).all() + +# Asignar al rol +for perm in invoice_perms: + role_perm = RolePermission( + client_role_id=role.id, + permission_id=perm.id + ) + db.add(role_perm) + +db.commit() +``` + +### Asignar rol a usuario + +```python +service.assign_role_to_user( + user_id="user-uuid-from-keycloak", + client_id=1, + role_id=role.id, + assigned_by="admin-uuid" +) +``` + +### Conceder permiso temporal + +```python +from datetime import datetime, timedelta + +service.grant_direct_permission( + user_id="user-uuid", + client_id=1, + permission_code="invoice.delete", + assigned_by="admin-uuid", + expires_at=datetime.utcnow() + timedelta(days=7) +) +``` + +## ⚡ Optimización de Rendimiento + +### 1. Caché con Redis + +```python +import redis +from functools import lru_cache + +redis_client = redis.Redis(host='localhost', port=6379) + +def get_cached_permissions(user_id: str, client_id: int) -> set: + cache_key = f"perms:{user_id}:{client_id}" + + cached = redis_client.get(cache_key) + if cached: + return set(cached.decode().split(',')) + + # Consultar DB + service = PermissionService(db) + permissions = service.get_user_permissions(user_id, client_id) + + # Cachear por 5 minutos + redis_client.setex(cache_key, 300, ','.join(permissions)) + + return permissions +``` + +### 2. Índices de Base de Datos + +Ya están definidos en los modelos: + +- Índices compuestos para consultas eficientes +- Índices únicos para prevenir duplicados +- Índices en foreign keys + +### 3. Query Optimization + +El servicio usa JOINs eficientes en lugar de N+1 queries. + +## 🧪 Testing + +```python +import pytest +from api.v1.modules.core.permissions import PermissionService +from api.v1.modules.core.permissions.models import Permission, ClientRole + +def test_user_has_permission_from_role(db_session): + # Setup + perm = Permission(code="invoice.view", module="invoice", action="view") + db_session.add(perm) + + role = ClientRole(client_id=1, code="viewer", name="Viewer") + db_session.add(role) + db_session.commit() + + # Test + service = PermissionService(db_session) + assert service.has_permission("user-123", 1, "invoice.view") +``` + +## 📝 Notas Importantes + +- **Client ID**: Por defecto se obtiene del header `X-Client-ID`, pero puede adaptarse a subdominios o JWT +- **User ID**: Se extrae del campo `sub` del token JWT de Keycloak +- **Permisos Directos**: Pueden revocar permisos heredados de roles (`is_granted=False`) +- **Soft Delete**: Los roles y permisos se desactivan (`is_active=False`) en lugar de eliminarse + +## 🔗 Integración con Keycloak + +Los roles globales de Keycloak pueden coexistir con los roles locales: + +```python +@router.get("/protected") +async def protected_route( + current_user: dict = Depends(get_current_user), + permissions: set = Depends(get_current_user_permissions) +): + # Verificar rol global de Keycloak + keycloak_roles = current_user.get("realm_access", {}).get("roles", []) + + if "super_admin" in keycloak_roles: + # Super admin tiene acceso total + return {"access": "granted", "level": "global"} + + # Verificar permisos a nivel de cliente + if "invoice.view" in permissions: + return {"access": "granted", "level": "client"} + + raise HTTPException(403, "No access") +``` diff --git a/backend/api/v1/modules/core/permissions/__init__.py b/backend/api/v1/modules/core/permissions/__init__.py new file mode 100644 index 00000000..06a6a509 --- /dev/null +++ b/backend/api/v1/modules/core/permissions/__init__.py @@ -0,0 +1,38 @@ +""" +Módulo de permisos multi-tenant. +Proporciona modelos, servicios y rutas para gestión de permisos granulares por companye. +""" + +from .models import ( + Permission, + CompanyRole, + RolePermission, + UserCompanyRole, + UserCompanyPermission, +) +from .service import PermissionService +from .dependencies import ( + PermissionChecker, + RequirePermission, + get_permission_service, + get_current_user_permissions, +) +from .routes import router + +__all__ = [ + # Models + "Permission", + "CompanyRole", + "RolePermission", + "UserCompanyRole", + "UserCompanyPermission", + # Service + "PermissionService", + # Dependencies + "PermissionChecker", + "RequirePermission", + "get_permission_service", + "get_current_user_permissions", + # Router + "router", +] diff --git a/backend/api/v1/modules/core/permissions/dependencies.py b/backend/api/v1/modules/core/permissions/dependencies.py new file mode 100644 index 00000000..95354071 --- /dev/null +++ b/backend/api/v1/modules/core/permissions/dependencies.py @@ -0,0 +1,231 @@ +""" +Dependencias de FastAPI para verificación de permisos multi-tenant. +Proporciona decoradores y funciones para proteger rutas con permisos específicos. +""" + +from typing import List, Optional, Callable +from fastapi import Depends, HTTPException, status, Header +from sqlalchemy.orm import Session +from functools import wraps +from core.database import get_core_db +from core.security import get_current_user # Asumiendo que existe esta función +from .service import PermissionService + + +# Dependencia para obtener el ID del cliente del header o contexto +async def get_client_id( + x_client_id: Optional[str] = Header(None, alias="X-Client-ID") +) -> int: + """ + Obtiene el ID del cliente desde el header de la petición. + + En producción, esto podría obtenerse de: + - Un header HTTP (X-Client-ID) + - Un subdomain (cliente1.miapp.com) + - El token JWT del usuario + - La sesión del usuario + """ + if not x_client_id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Client ID is required. Provide X-Client-ID header.", + ) + + try: + return int(x_client_id) + except ValueError: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid client ID format" + ) + + +# Dependencia para obtener el servicio de permisos +def get_permission_service(db: Session = Depends(get_core_db)) -> PermissionService: + """ + Crea una instancia del servicio de permisos con la sesión de base de datos. + """ + return PermissionService(db) + + +# Clase para verificación de permisos (puede usarse como dependencia) +class PermissionChecker: + """ + Verificador de permisos que puede usarse como dependencia de FastAPI. + + Ejemplo de uso: + @app.get("/invoices") + async def list_invoices( + _: None = Depends(PermissionChecker(["invoice.view"])) + ): + return {"invoices": [...]} + """ + + def __init__(self, required_permissions: List[str], require_all: bool = True): + """ + Args: + required_permissions: Lista de permisos requeridos + require_all: Si True, requiere TODOS los permisos. + Si False, requiere AL MENOS UNO. + """ + self.required_permissions = required_permissions + self.require_all = require_all + + async def __call__( + self, + current_user: dict = Depends(get_current_user), + client_id: int = Depends(get_client_id), + permission_service: PermissionService = Depends(get_permission_service), + ): + """ + Verifica que el usuario tenga los permisos requeridos. + + Lanza HTTPException 403 si no tiene permisos. + """ + user_id = current_user.get("sub") or current_user.get("id") + + if not user_id: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="User ID not found in token", + ) + + # Verificar permisos + if self.require_all: + has_access = permission_service.has_all_permissions( + user_id=user_id, + client_id=client_id, + permission_codes=self.required_permissions, + ) + else: + has_access = permission_service.has_any_permission( + user_id=user_id, + client_id=client_id, + permission_codes=self.required_permissions, + ) + + if not has_access: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Missing required permissions: {', '.join(self.required_permissions)}", + ) + + return True + + +# Función alternativa para verificar un solo permiso +class RequirePermission: + """ + Verificador simple para un único permiso. + + Ejemplo: + @app.post("/invoices") + async def create_invoice( + _: None = Depends(RequirePermission("invoice.create")) + ): + return {"created": True} + """ + + def __init__(self, permission_code: str): + self.permission_code = permission_code + + async def __call__( + self, + current_user: dict = Depends(get_current_user), + client_id: int = Depends(get_client_id), + permission_service: PermissionService = Depends(get_permission_service), + ): + user_id = current_user.get("sub") or current_user.get("id") + + if not user_id: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="User ID not found in token", + ) + + has_permission = permission_service.has_permission( + user_id=user_id, client_id=client_id, permission_code=self.permission_code + ) + + if not has_permission: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Missing required permission: {self.permission_code}", + ) + + return True + + +# Decorador personalizado para aplicar a funciones (opcional) +def require_permissions(*permissions: str, require_all: bool = True): + """ + Decorador para verificar permisos en funciones. + Útil para lógica de negocio fuera de rutas FastAPI. + + Ejemplo: + @require_permissions("invoice.edit", "invoice.view") + def update_invoice_logic(invoice_id: int, user_id: str, client_id: int, db: Session): + # Lógica de actualización + pass + """ + + def decorator(func: Callable): + @wraps(func) + def wrapper(*args, **kwargs): + # Extraer user_id, client_id y db de los argumentos + user_id = kwargs.get("user_id") + client_id = kwargs.get("client_id") + db = kwargs.get("db") + + if not all([user_id, client_id, db]): + raise ValueError( + "Function must receive 'user_id', 'client_id', and 'db' as keyword arguments" + ) + + # Verificar permisos + permission_service = PermissionService(db) + + if require_all: + has_access = permission_service.has_all_permissions( + user_id=user_id, + client_id=client_id, + permission_codes=list(permissions), + ) + else: + has_access = permission_service.has_any_permission( + user_id=user_id, + client_id=client_id, + permission_codes=list(permissions), + ) + + if not has_access: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail=f"Missing required permissions: {', '.join(permissions)}", + ) + + return func(*args, **kwargs) + + return wrapper + + return decorator + + +# Función helper para obtener permisos del usuario actual +async def get_current_user_permissions( + current_user: dict = Depends(get_current_user), + client_id: int = Depends(get_client_id), + permission_service: PermissionService = Depends(get_permission_service), +) -> set: + """ + Devuelve todos los permisos del usuario actual en el cliente. + Útil para endpoints que necesitan conocer los permisos disponibles. + """ + user_id = current_user.get("sub") or current_user.get("id") + + if not user_id: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="User ID not found in token", + ) + + return permission_service.get_user_permissions(user_id, client_id) diff --git a/backend/api/v1/modules/core/permissions/integration_example.py b/backend/api/v1/modules/core/permissions/integration_example.py new file mode 100644 index 00000000..e368e884 --- /dev/null +++ b/backend/api/v1/modules/core/permissions/integration_example.py @@ -0,0 +1,153 @@ +""" +Ejemplo de integración del módulo de permisos en el router principal. +Este archivo muestra cómo incluir las rutas de permisos en tu API. +""" + +from fastapi import APIRouter, Depends +from api.v1.modules.core.permissions import router as permissions_router + +# Crear el router principal de la API v1 +api_v1_router = APIRouter(prefix="/api/v1") + +# ============================================================================ +# REGISTRAR EL MÓDULO DE PERMISOS +# ============================================================================ + +# Opción 1: Incluir directamente con el prefix ya definido en el módulo +api_v1_router.include_router(permissions_router) +# Esto creará rutas como: /api/v1/permissions/me, /api/v1/permissions/roles, etc. + + +# Opción 2: Sobrescribir el prefix si lo necesitas +# api_v1_router.include_router( +# permissions_router, +# prefix="/permisos", # Cambiar a español si prefieres +# tags=["Permisos"] +# ) + + +# ============================================================================ +# IMPORTAR DEPENDENCIAS PARA USAR EN OTRAS RUTAS +# ============================================================================ + +from api.v1.modules.core.permissions import ( + RequirePermission, + PermissionChecker, + get_current_user_permissions, + PermissionService +) + +# ============================================================================ +# EJEMPLO: USAR PERMISOS EN OTROS MÓDULOS +# ============================================================================ + +# Supongamos que tienes un módulo de facturas +invoices_router = APIRouter(prefix="/invoices", tags=["Invoices"]) + +@invoices_router.get("/") +async def list_invoices( + _: None = Depends(RequirePermission("invoice.view")) +): + """Lista todas las facturas (requiere permiso invoice.view).""" + return { + "invoices": [ + {"id": 1, "number": "INV-001", "amount": 1000}, + {"id": 2, "number": "INV-002", "amount": 2000} + ] + } + + +@invoices_router.post("/") +async def create_invoice( + _: None = Depends(PermissionChecker(["invoice.view", "invoice.create"], require_all=True)) +): + """Crea una factura (requiere invoice.view + invoice.create).""" + return {"message": "Invoice created", "id": 3} + + +@invoices_router.delete("/{invoice_id}") +async def delete_invoice( + invoice_id: int, + _: None = Depends(RequirePermission("invoice.delete")) +): + """Elimina una factura (requiere invoice.delete).""" + return {"message": f"Invoice {invoice_id} deleted"} + + +# Registrar el router de facturas +api_v1_router.include_router(invoices_router) + + +# ============================================================================ +# EJEMPLO: RUTA CON LÓGICA CONDICIONAL SEGÚN PERMISOS +# ============================================================================ + +reports_router = APIRouter(prefix="/reports", tags=["Reports"]) + +@reports_router.get("/dashboard") +async def get_reports_dashboard( + user_permissions: set = Depends(get_current_user_permissions) +): + """ + Dashboard de reportes que muestra diferentes secciones según los permisos. + No requiere permisos específicos, pero el contenido varía. + """ + sections = [] + + if "report.financial.view" in user_permissions: + sections.append({ + "type": "financial", + "title": "Reportes Financieros", + "reports": ["Balance General", "Estado de Resultados", "Flujo de Efectivo"] + }) + + if "report.admin.view" in user_permissions: + sections.append({ + "type": "administrative", + "title": "Reportes Administrativos", + "reports": ["Actividad de Usuarios", "Auditoría", "Logs del Sistema"] + }) + + if "invoice.view" in user_permissions: + sections.append({ + "type": "invoices", + "title": "Reportes de Facturación", + "reports": ["Facturas por Mes", "Facturas Pendientes", "Análisis de Ventas"] + }) + + return { + "user_permissions": list(user_permissions), + "available_sections": sections + } + + +# Registrar el router de reportes +api_v1_router.include_router(reports_router) + + +# ============================================================================ +# EXPORTAR EL ROUTER PRINCIPAL +# ============================================================================ + +__all__ = ["api_v1_router"] + + +# ============================================================================ +# USO EN main.py +# ============================================================================ +""" +En tu backend/main.py, incluye el router así: + +from fastapi import FastAPI +from api.v1.integration_example import api_v1_router + +app = FastAPI(title="Mi API Multi-Tenant") + +# Registrar todas las rutas de API v1 +app.include_router(api_v1_router) + +# Ahora tendrás: +# - /api/v1/permissions/* (rutas de permisos) +# - /api/v1/invoices/* (rutas de facturas protegidas) +# - /api/v1/reports/* (rutas de reportes) +""" diff --git a/backend/api/v1/modules/core/permissions/models.py b/backend/api/v1/modules/core/permissions/models.py new file mode 100644 index 00000000..03ca0ddb --- /dev/null +++ b/backend/api/v1/modules/core/permissions/models.py @@ -0,0 +1,247 @@ +""" +Modelos de permisos multi-tenant para el sistema. +Este módulo define el sistema de permisos granular por compañia/tenant. +""" + +from datetime import datetime, timezone +from typing import Optional +from sqlalchemy import ( + String, + Integer, + ForeignKey, + DateTime, + Boolean, + UniqueConstraint, + Index, +) +from sqlalchemy.orm import relationship, Mapped, mapped_column +from api.v1.common.base_models import TenantScopedMixin, TimestampMixin +from core.database import Base + + +# Modelo para permisos del sistema +# Representa acciones específicas como "invoice.view", "invoice.edit", etc. +class Permission(Base, TimestampMixin): + __tablename__ = "permissions" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + code: Mapped[str] = mapped_column( + String(100), unique=True, nullable=False, index=True + ) + # Código único del permiso (ej: "invoice.view", "user.edit") + + description: Mapped[Optional[str]] = mapped_column(String(255)) + # Descripción legible del permiso + + module: Mapped[str] = mapped_column(String(50), nullable=False, index=True) + # Módulo al que pertenece (ej: "invoice", "user", "report") + + action: Mapped[str] = mapped_column(String(50), nullable=False) + # Acción específica (ej: "view", "edit", "delete", "create") + + is_active: Mapped[bool] = mapped_column( + Boolean, default=True, server_default="true", nullable=False + ) + # Permite desactivar permisos sin eliminarlos + + __table_args__ = {"schema": "core", "extend_existing": True} + + # Relaciones + role_permissions: Mapped[list["RolePermission"]] = relationship( + "RolePermission", back_populates="permission", cascade="all, delete-orphan" + ) + user_company_permissions: Mapped[list["UserCompanyPermission"]] = relationship( + "UserCompanyPermission", + back_populates="permission", + cascade="all, delete-orphan", + ) + + +# Modelo para roles personalizados por compañia/tenant +# Cada compañia puede definir sus propios roles con nombres personalizados +class CompanyRole(Base, TenantScopedMixin, TimestampMixin): + __tablename__ = "company_roles" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + + name: Mapped[str] = mapped_column(String(100), nullable=False) + # Nombre del rol (ej: "Administrador", "Contador", "Vendedor") + + code: Mapped[str] = mapped_column(String(100), nullable=False) + # Código único del rol dentro del compañia (ej: "admin", "accountant") + + description: Mapped[Optional[str]] = mapped_column(String(255)) + # Descripción del rol + + is_active: Mapped[bool] = mapped_column( + Boolean, default=True, server_default="true", nullable=False + ) + # Permite desactivar roles sin eliminarlos + + # Restricción: el código del rol debe ser único por compañia + __table_args__ = ( + UniqueConstraint( + "company_id", "tenant_id", "code", name="uq_company_role_code" + ), + Index( + "ix_company_roles_company_id_is_active", + "company_id", + "tenant_id", + "is_active", + ), + {"schema": "core", "extend_existing": True}, + ) + + # Relaciones + role_permissions: Mapped[list["RolePermission"]] = relationship( + "RolePermission", back_populates="company_role", cascade="all, delete-orphan" + ) + user_company_roles: Mapped[list["UserCompanyRole"]] = relationship( + "UserCompanyRole", + back_populates="company_role", + cascade="all, delete-orphan", + ) + + +# Tabla de relación entre roles de compañia y permisos +# Define qué permisos tiene cada rol +class RolePermission(Base, TenantScopedMixin, TimestampMixin): + __tablename__ = "role_permissions" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + + company_role_id: Mapped[int] = mapped_column( + Integer, + ForeignKey("core.company_roles.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + + permission_id: Mapped[int] = mapped_column( + Integer, + ForeignKey("core.permissions.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + + # Restricción: un permiso no puede estar duplicado en el mismo rol + __table_args__ = ( + UniqueConstraint("company_role_id", "permission_id", name="uq_role_permission"), + Index("ix_role_permissions_composite", "company_role_id", "permission_id"), + {"schema": "core", "extend_existing": True}, + ) + + # Relaciones + company_role: Mapped["CompanyRole"] = relationship( + "CompanyRole", back_populates="role_permissions" + ) + permission: Mapped["Permission"] = relationship( + "Permission", back_populates="role_permissions" + ) + + +# Tabla de relación entre usuarios y roles de compañia +# Define qué roles tiene cada usuario en cada compañia +class UserCompanyRole(Base, TenantScopedMixin, TimestampMixin): + __tablename__ = "user_company_roles" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + + user_id: Mapped[str] = mapped_column(String(100), nullable=False, index=True) + # ID del usuario (puede ser UUID de Keycloak u otro identificador) + + company_role_id: Mapped[int] = mapped_column( + Integer, + ForeignKey("core.company_roles.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + + is_active: Mapped[bool] = mapped_column( + Boolean, default=True, server_default="true", nullable=False + ) + # Permite desactivar asignaciones sin eliminarlas + + assigned_by: Mapped[Optional[str]] = mapped_column(String(100)) + # ID del usuario que asignó este rol + + # Restricción: un usuario no puede tener el mismo rol duplicado en un compañia + __table_args__ = ( + UniqueConstraint( + "user_id", + "company_id", + "tenant_id", + "company_role_id", + name="uq_user_company_role", + ), + Index( + "ix_user_company_roles_user_company", + "user_id", + "company_id", + "tenant_id", + "is_active", + ), + {"schema": "core", "extend_existing": True}, + ) + + # Relaciones + company_role: Mapped["CompanyRole"] = relationship( + "CompanyRole", back_populates="user_company_roles" + ) + + +# Tabla para permisos directos de usuario por compañia (opcional) +# Permite asignar permisos específicos a un usuario sin necesidad de un rol +# Útil para casos excepcionales o permisos temporales +class UserCompanyPermission(Base, TenantScopedMixin, TimestampMixin): + __tablename__ = "user_company_permissions" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True) + + user_id: Mapped[str] = mapped_column(String(100), nullable=False, index=True) + # ID del usuario + + permission_id: Mapped[int] = mapped_column( + Integer, + ForeignKey("core.permissions.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + + is_granted: Mapped[bool] = mapped_column( + Boolean, default=True, server_default="true", nullable=False + ) + # True = permiso concedido, False = permiso revocado explícitamente + # Permite revocar permisos que vienen de roles + + is_active: Mapped[bool] = mapped_column( + Boolean, default=True, server_default="true", nullable=False + ) + + assigned_by: Mapped[Optional[str]] = mapped_column(String(100)) + expires_at: Mapped[Optional[datetime]] = mapped_column(DateTime) + # Permite permisos temporales con fecha de expiración + + # Restricción: un usuario no puede tener el mismo permiso duplicado en un compañia + __table_args__ = ( + UniqueConstraint( + "user_id", + "company_id", + "tenant_id", + "permission_id", + name="uq_user_company_permission", + ), + Index( + "ix_user_company_permissions_composite", + "user_id", + "company_id", + "tenant_id", + "is_active", + ), + {"schema": "core", "extend_existing": True}, + ) + + # Relaciones + permission: Mapped["Permission"] = relationship( + "Permission", back_populates="user_company_permissions" + ) diff --git a/backend/api/v1/modules/core/permissions/routes.py b/backend/api/v1/modules/core/permissions/routes.py new file mode 100644 index 00000000..2fa738b0 --- /dev/null +++ b/backend/api/v1/modules/core/permissions/routes.py @@ -0,0 +1,445 @@ +""" +Rutas API para el sistema de permisos multi-tenant. +Endpoints para gestión de permisos, roles y asignaciones. +""" + +from typing import List, Optional, Dict, Any +from fastapi import APIRouter, Depends, HTTPException, status, Query +from sqlalchemy.orm import Session + +from core.database import get_core_db +from core.security import get_current_user, validate_access_to_resource +from .dependencies import ( + PermissionChecker, + RequirePermission, + get_permission_service, + get_current_user_permissions, +) +from .service import PermissionService +from .models import Permission, CompanyRole +from .schemas import ( + PermissionResponse, + CompanyRoleResponse, + UserPermissionsResponse, + AssignRoleRequest, + GrantPermissionRequest, + SuccessResponse, + CreateRoleRequest, + UpdateRoleRequest, + AssignPermissionsToRoleRequest, + PermissionListResponse, + RoleListResponse, +) + + +router = APIRouter(prefix="/permissions", tags=["permissions"]) + + +# ============================================================================ +# RUTAS DE CONSULTA DE PERMISOS +# ============================================================================ + + +@router.get("/me", response_model=UserPermissionsResponse) +async def get_my_permissions( + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), + permission_service: PermissionService = Depends(get_permission_service), +): + """ + Obtiene los permisos y roles del usuario actual en el company actual. + No requiere permisos especiales ya que es información propia. + """ + tenant_id = validate_access_to_resource(db, company_id, current_user) + + user_id = current_user.get("sub") or current_user.get("id") + + # Obtener permisos + permissions = permission_service.get_user_permissions(user_id, company_id) + + # Obtener roles + roles = permission_service.get_user_roles(user_id, company_id) + role_names = [role.name for role in roles] + + return UserPermissionsResponse( + user_id=user_id, + company_id=company_id, + permissions=list(permissions), + roles=role_names, + ) + + +@router.get("/available", response_model=PermissionListResponse) +async def list_available_permissions( + db: Session = Depends(get_core_db), + page: int = Query(1, ge=1, description="Número de página"), + page_size: int = Query(100, ge=1, le=500, description="Tamaño de página"), + module: Optional[str] = Query(None, description="Filtrar por módulo"), + _: None = Depends(RequirePermission("permissions.view")), +): + """ + Lista todos los permisos disponibles en el sistema. + Requiere permiso: permissions.view + """ + query = db.query(Permission).filter(Permission.is_active == True) + + if module: + query = query.filter(Permission.module == module) + + total = query.count() + permissions = query.offset((page - 1) * page_size).limit(page_size).all() + + return PermissionListResponse( + items=permissions, total=total, page=page, page_size=page_size + ) + + +@router.get("/roles", response_model=RoleListResponse) +async def list_company_roles( + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), + page: int = Query(1, ge=1, description="Número de página"), + page_size: int = Query(100, ge=1, le=500, description="Tamaño de página"), + _: None = Depends(RequirePermission("roles.view")), +): + """ + Lista todos los roles del companye actual. + Requiere permiso: roles.view + """ + tenant_id = validate_access_to_resource(db, company_id, current_user) + + query = db.query(CompanyRole).filter( + CompanyRole.company_id == company_id, CompanyRole.is_active == True + ) + + total = query.count() + roles = query.offset((page - 1) * page_size).limit(page_size).all() + + return RoleListResponse(items=roles, total=total, page=page, page_size=page_size) + + +@router.get("/users/{user_id}", response_model=UserPermissionsResponse) +async def get_user_permissions( + user_id: str, + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), + permission_service: PermissionService = Depends(get_permission_service), + _: None = Depends(RequirePermission("user.view")), +): + """ + Obtiene los permisos y roles de un usuario específico. + Requiere permiso: user.view + """ + tenant_id = validate_access_to_resource(db, company_id, current_user) + + permissions = permission_service.get_user_permissions(user_id, company_id) + roles = permission_service.get_user_roles(user_id, company_id) + role_names = [role.name for role in roles] + + return UserPermissionsResponse( + user_id=user_id, + company_id=company_id, + permissions=list(permissions), + roles=role_names, + ) + + +# ============================================================================ +# RUTAS DE GESTIÓN DE ROLES +# ============================================================================ + + +@router.post( + "/roles", response_model=CompanyRoleResponse, status_code=status.HTTP_201_CREATED +) +async def create_role( + request: CreateRoleRequest, + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), + _: None = Depends(RequirePermission("roles.create")), +): + """ + Crea un nuevo rol personalizado para el companye. + Requiere permiso: roles.create + """ + tenant_id = validate_access_to_resource(db, company_id, current_user) + + # Verificar que el código no esté en uso + existing = ( + db.query(CompanyRole) + .filter(CompanyRole.company_id == company_id, CompanyRole.code == request.code) + .first() + ) + + if existing: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=f"Role with code '{request.code}' already exists for this company", + ) + + # Crear el rol + role = CompanyRole( + company_id=company_id, + name=request.name, + code=request.code, + description=request.description, + ) + + db.add(role) + db.commit() + db.refresh(role) + + # Asignar permisos si se especificaron + if request.permission_ids: + from .models import RolePermission + + for perm_id in request.permission_ids: + role_perm = RolePermission(company_role_id=role.id, permission_id=perm_id) + db.add(role_perm) + db.commit() + + return role + + +@router.patch("/roles/{role_id}", response_model=CompanyRoleResponse) +async def update_role( + role_id: int, + request: UpdateRoleRequest, + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), + _: None = Depends(RequirePermission("roles.edit")), +): + """ + Actualiza un rol existente. + Requiere permiso: roles.edit + """ + tenant_id = validate_access_to_resource(db, company_id, current_user) + + role = ( + db.query(CompanyRole) + .filter(CompanyRole.id == role_id, CompanyRole.company_id == company_id) + .first() + ) + + if not role: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Role not found" + ) + + # Actualizar campos + if request.name is not None: + role.name = request.name + if request.description is not None: + role.description = request.description + if request.is_active is not None: + role.is_active = request.is_active + + db.commit() + db.refresh(role) + + return role + + +@router.delete("/roles/{role_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_role( + role_id: int, + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), + _: None = Depends(RequirePermission("roles.delete")), +): + """ + Elimina (desactiva) un rol. + Requiere permiso: roles.delete + """ + tenant_id = validate_access_to_resource(db, company_id, current_user) + + role = ( + db.query(CompanyRole) + .filter(CompanyRole.id == role_id, CompanyRole.company_id == company_id) + .first() + ) + + if not role: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Role not found" + ) + + role.is_active = False + db.commit() + + +# ============================================================================ +# RUTAS DE ASIGNACIÓN DE ROLES Y PERMISOS +# ============================================================================ + + +@router.post( + "/assign-role", response_model=SuccessResponse, status_code=status.HTTP_201_CREATED +) +async def assign_role( + request: AssignRoleRequest, + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), + permission_service: PermissionService = Depends(get_permission_service), + _: None = Depends(RequirePermission("roles.assign")), +): + """ + Asigna un rol a un usuario en el companye actual. + Requiere permiso: roles.assign + """ + tenant_id = validate_access_to_resource(db, company_id, current_user) + + assigner_id = current_user.get("sub") or current_user.get("id") + + try: + user_role = permission_service.assign_role_to_user( + user_id=request.user_id, + company_id=company_id, + role_id=request.role_id, + assigned_by=assigner_id, + ) + + return SuccessResponse( + message=f"Role assigned to user {request.user_id}", + data={"assignment_id": user_role.id}, + ) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) + + +@router.post( + "/grant-permission", + response_model=SuccessResponse, + status_code=status.HTTP_201_CREATED, +) +async def grant_permission( + request: GrantPermissionRequest, + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), + permission_service: PermissionService = Depends(get_permission_service), + _: None = Depends(RequirePermission("permissions.grant")), +): + """ + Concede un permiso directo a un usuario en el companye actual. + Requiere permiso: permissions.grant + """ + tenant_id = validate_access_to_resource(db, company_id, current_user) + + assigner_id = current_user.get("sub") or current_user.get("id") + + try: + user_permission = permission_service.grant_direct_permission( + user_id=request.user_id, + company_id=company_id, + permission_code=request.permission_code, + assigned_by=assigner_id, + expires_at=request.expires_at, + ) + + return SuccessResponse( + message=f"Permission {request.permission_code} granted to user {request.user_id}", + data={ + "permission_id": user_permission.id, + "expires_at": ( + user_permission.expires_at.isoformat() + if user_permission.expires_at + else None + ), + }, + ) + except ValueError as e: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(e)) + + +# ============================================================================ +# EJEMPLOS DE RUTAS PROTEGIDAS CON PERMISOS +# ============================================================================ + + +@router.get("/examples/invoices") +async def example_view_invoices(_: None = Depends(RequirePermission("invoice.view"))): + """ + Endpoint de ejemplo que requiere el permiso invoice.view + """ + return { + "message": "You have permission to view invoices", + "invoices": [ + {"id": 1, "number": "INV-001", "amount": 1000}, + {"id": 2, "number": "INV-002", "amount": 2000}, + ], + } + + +@router.post("/examples/invoices") +async def example_edit_invoice( + _: None = Depends( + PermissionChecker(["invoice.view", "invoice.edit"], require_all=True) + ) +): + """ + Endpoint de ejemplo que requiere AMBOS permisos: invoice.view e invoice.edit + """ + return { + "message": "Invoice updated successfully", + "invoice": {"id": 1, "number": "INV-001", "amount": 1500}, + } + + +@router.get("/examples/reports") +async def example_view_reports( + _: None = Depends( + PermissionChecker( + ["report.financial.view", "report.admin.view"], require_all=False + ) + ) +): + """ + Endpoint de ejemplo que requiere AL MENOS UNO de los permisos especificados. + """ + return { + "message": "Financial reports", + "reports": ["Monthly P&L", "Cash Flow", "Balance Sheet"], + } + + +@router.get("/examples/dashboard") +async def example_dashboard( + user_permissions: set = Depends(get_current_user_permissions), +): + """ + Dashboard dinámico que muestra diferentes widgets según los permisos del usuario. + """ + widgets = [] + + if "invoice.view" in user_permissions: + widgets.append( + { + "type": "invoices", + "title": "Recent Invoices", + "data": [{"id": 1, "number": "INV-001"}], + } + ) + + if "report.financial.view" in user_permissions: + widgets.append( + { + "type": "financial", + "title": "Financial Summary", + "data": {"revenue": 50000, "expenses": 30000}, + } + ) + + if "user.view" in user_permissions: + widgets.append( + {"type": "users", "title": "User Activity", "data": {"active_users": 42}} + ) + + return {"permissions": list(user_permissions), "widgets": widgets} diff --git a/backend/api/v1/modules/core/permissions/schemas.py b/backend/api/v1/modules/core/permissions/schemas.py new file mode 100644 index 00000000..36f13aec --- /dev/null +++ b/backend/api/v1/modules/core/permissions/schemas.py @@ -0,0 +1,245 @@ +""" +Esquemas Pydantic para el módulo de permisos. +Define los modelos de request/response para las APIs de permisos. +""" + +from typing import List, Optional +from datetime import datetime +from pydantic import BaseModel, Field, ConfigDict + + +# ============================================================================ +# SCHEMAS DE RESPONSE +# ============================================================================ + + +class PermissionResponse(BaseModel): + """Esquema de respuesta para un permiso individual.""" + + id: int + code: str = Field(..., description="Código único del permiso (ej: 'invoice.view')") + description: Optional[str] = Field(None, description="Descripción del permiso") + module: str = Field(..., description="Módulo al que pertenece (ej: 'invoice')") + action: str = Field(..., description="Acción específica (ej: 'view', 'edit')") + is_active: bool = Field(..., description="Si el permiso está activo") + + model_config = ConfigDict(from_attributes=True) + + +class CompanyRoleResponse(BaseModel): + """Esquema de respuesta para un rol de companye.""" + + id: int + company_id: int = Field(..., description="ID del companye al que pertenece el rol") + name: str = Field(..., description="Nombre del rol (ej: 'Administrador')") + code: str = Field(..., description="Código del rol (ej: 'admin')") + description: Optional[str] = Field(None, description="Descripción del rol") + is_active: bool = Field(..., description="Si el rol está activo") + created_at: datetime + updated_at: datetime + + model_config = ConfigDict(from_attributes=True) + + +class CompanyRoleWithPermissionsResponse(CompanyRoleResponse): + """Esquema de respuesta para un rol con sus permisos incluidos.""" + + permissions: List[PermissionResponse] = Field( + default_factory=list, description="Lista de permisos asignados a este rol" + ) + + +class UserPermissionsResponse(BaseModel): + """Esquema de respuesta para los permisos de un usuario.""" + + user_id: str = Field(..., description="ID del usuario") + company_id: int = Field(..., description="ID del companye") + permissions: List[str] = Field( + default_factory=list, description="Lista de códigos de permisos del usuario" + ) + roles: List[str] = Field( + default_factory=list, description="Lista de nombres de roles del usuario" + ) + + +class UserCompanyRoleResponse(BaseModel): + """Esquema de respuesta para la asignación de rol a usuario.""" + + id: int + user_id: str + company_id: int + company_role_id: int + is_active: bool + assigned_at: datetime + assigned_by: Optional[str] = None + + model_config = ConfigDict(from_attributes=True) + + +class UserCompanyPermissionResponse(BaseModel): + """Esquema de respuesta para un permiso directo de usuario.""" + + id: int + user_id: str + company_id: int + permission_id: int + permission_code: Optional[str] = None + is_granted: bool = Field( + ..., description="True si está concedido, False si está revocado" + ) + is_active: bool + assigned_at: datetime + assigned_by: Optional[str] = None + expires_at: Optional[datetime] = Field( + None, description="Fecha de expiración del permiso" + ) + + model_config = ConfigDict(from_attributes=True) + + +# ============================================================================ +# SCHEMAS DE REQUEST +# ============================================================================ + + +class AssignRoleRequest(BaseModel): + """Esquema de request para asignar un rol a un usuario.""" + + user_id: str = Field(..., description="ID del usuario al que se asignará el rol") + role_id: int = Field(..., description="ID del rol a asignar") + + +class RemoveRoleRequest(BaseModel): + """Esquema de request para remover un rol de un usuario.""" + + user_id: str = Field(..., description="ID del usuario") + role_id: int = Field(..., description="ID del rol a remover") + + +class GrantPermissionRequest(BaseModel): + """Esquema de request para conceder un permiso directo a un usuario.""" + + user_id: str = Field(..., description="ID del usuario") + permission_code: str = Field( + ..., description="Código del permiso a conceder (ej: 'invoice.delete')" + ) + expires_at: Optional[datetime] = Field( + None, description="Fecha de expiración del permiso (opcional)" + ) + + +class RevokePermissionRequest(BaseModel): + """Esquema de request para revocar un permiso directo.""" + + user_id: str = Field(..., description="ID del usuario") + permission_code: str = Field(..., description="Código del permiso a revocar") + + +class CreateRoleRequest(BaseModel): + """Esquema de request para crear un rol personalizado.""" + + name: str = Field(..., min_length=1, max_length=100, description="Nombre del rol") + code: str = Field( + ..., + min_length=1, + max_length=100, + description="Código único del rol (ej: 'custom_admin')", + ) + description: Optional[str] = Field( + None, max_length=255, description="Descripción del rol" + ) + permission_ids: List[int] = Field( + default_factory=list, description="IDs de permisos a asignar al rol" + ) + + +class UpdateRoleRequest(BaseModel): + """Esquema de request para actualizar un rol existente.""" + + name: Optional[str] = Field(None, min_length=1, max_length=100) + description: Optional[str] = Field(None, max_length=255) + is_active: Optional[bool] = None + + +class AssignPermissionsToRoleRequest(BaseModel): + """Esquema de request para asignar permisos a un rol.""" + + permission_ids: List[int] = Field( + ..., description="Lista de IDs de permisos a asignar al rol" + ) + replace_existing: bool = Field( + False, + description="Si True, reemplaza los permisos existentes. Si False, los agrega.", + ) + + +class CreatePermissionRequest(BaseModel): + """Esquema de request para crear un nuevo permiso (uso administrativo).""" + + code: str = Field( + ..., + min_length=1, + max_length=100, + description="Código único del permiso (ej: 'custom_module.action')", + ) + description: Optional[str] = Field(None, max_length=255) + module: str = Field( + ..., min_length=1, max_length=50, description="Módulo del permiso" + ) + action: str = Field( + ..., min_length=1, max_length=50, description="Acción del permiso" + ) + + +# ============================================================================ +# SCHEMAS DE RESPUESTA GENÉRICOS +# ============================================================================ + + +class SuccessResponse(BaseModel): + """Respuesta genérica de éxito.""" + + success: bool = True + message: str = Field(..., description="Mensaje descriptivo de la operación") + data: Optional[dict] = Field(None, description="Datos adicionales opcionales") + + +class ErrorResponse(BaseModel): + """Respuesta genérica de error.""" + + success: bool = False + detail: str = Field(..., description="Descripción del error") + error_code: Optional[str] = Field(None, description="Código de error específico") + + +# ============================================================================ +# SCHEMAS DE PAGINACIÓN +# ============================================================================ + + +class PaginatedResponse(BaseModel): + """Esquema genérico para respuestas paginadas.""" + + items: List[dict] = Field(default_factory=list) + total: int = Field(..., description="Total de items disponibles") + page: int = Field(..., description="Página actual") + page_size: int = Field(..., description="Tamaño de página") + total_pages: int = Field(..., description="Total de páginas disponibles") + + +class PermissionListResponse(BaseModel): + """Lista paginada de permisos.""" + + items: List[PermissionResponse] + total: int + page: int = 1 + page_size: int = 100 + + +class RoleListResponse(BaseModel): + """Lista paginada de roles.""" + + items: List[CompanyRoleResponse] + total: int + page: int = 1 + page_size: int = 100 diff --git a/backend/api/v1/modules/core/permissions/seed.py b/backend/api/v1/modules/core/permissions/seed.py new file mode 100644 index 00000000..2f393dbc --- /dev/null +++ b/backend/api/v1/modules/core/permissions/seed.py @@ -0,0 +1,63 @@ +seed_invoices = [ + ("invoice.imp.tem.view", "Ver facturas temporales de importación", "invoice", "view"), + ("invoice.imp.tem.create", "Crear facturas temporales de importación", "invoice", "create"), + ("invoice.imp.tem.edit", "Editar facturas", "invoice", "edit"), + ("invoice.imp.tem.delete", "Eliminar facturas", "invoice", "delete"), + ("invoice.imp.tem.process", "Procesar facturas", "invoice", "process"), + + ("invoice.imp.def.view", "Ver facturas definitivas de importación", "invoice", "view"), + ("invoice.imp.def.create", "Crear facturas definitivas de importación", "invoice", "create"), + ("invoice.imp.def.edit", "Editar facturas", "invoice", "edit"), + ("invoice.imp.def.delete", "Eliminar facturas", "invoice", "delete"), + ("invoice.imp.def.process", "Procesar facturas", "invoice", "process"), + + ("invoice.imp.cm.view", "Ver facturas temporales de importación", "invoice", "view"), + ("invoice.imp.cm.create", "Crear facturas temporales de importación", "invoice", "create"), + ("invoice.imp.cm.edit", "Editar facturas", "invoice", "edit"), + ("invoice.imp.cm.delete", "Eliminar facturas", "invoice", "delete"), + ("invoice.imp.cm.process", "Procesar facturas", "invoice", "process"), + + ("invoice.imp.cr.view", "Ver facturas temporales de importación", "invoice", "view"), + ("invoice.imp.cr.create", "Crear facturas temporales de importación", "invoice", "create"), + ("invoice.imp.cr.edit", "Editar facturas", "invoice", "edit"), + ("invoice.imp.cr.delete", "Eliminar facturas", "invoice", "delete"), + ("invoice.imp.cr.process", "Procesar facturas", "invoice", "process"), + + ("invoice.exp.view", "Ver facturas temporales de importación", "invoice", "view"), + ("invoice.exp.create", "Crear facturas temporales de importación", "invoice", "create"), + ("invoice.exp.edit", "Editar facturas", "invoice", "edit"), + ("invoice.exp.delete", "Eliminar facturas", "invoice", "delete"), + ("invoice.exp.process", "Procesar facturas", "invoice", "process"), + + ("invoice.exp.rep.view", "Ver facturas temporales de importación", "invoice", "view"), + ("invoice.exp.rep.create", "Crear facturas temporales de importación", "invoice", "create"), + ("invoice.exp.rep.edit", "Editar facturas", "invoice", "edit"), + ("invoice.exp.rep.delete", "Eliminar facturas", "invoice", "delete"), + ("invoice.exp.rep.process", "Procesar facturas", "invoice", "process"), + + +] + +seed_user = [ + ("user.view", "Ver facturas temporales de importación", "invoice", "view"), + ("user.create", "Crear facturas temporales de importación", "invoice", "create"), + ("user.edit", "Editar facturas", "invoice", "edit"), + ("user.delete", "Eliminar facturas", "invoice", "delete"), + ("user.process", "Procesar facturas", "invoice", "process"), +] + +seed_report = [ + ("report.view", "Ver facturas temporales de importación", "invoice", "view"), + ("report.create", "Crear facturas temporales de importación", "invoice", "create"), + ("report.edit", "Editar facturas", "invoice", "edit"), + ("report.delete", "Eliminar facturas", "invoice", "delete"), + ("report.process", "Procesar facturas", "invoice", "process"), +] + +seed_roles = [ + ("roles.view", "Ver facturas temporales de importación", "invoice", "view"), + ("roles.create", "Crear facturas temporales de importación", "invoice", "create"), + ("roles.edit", "Editar facturas", "invoice", "edit"), + ("roles.delete", "Eliminar facturas", "invoice", "delete"), + ("roles.process", "Procesar facturas", "invoice", "process"), +] \ No newline at end of file diff --git a/backend/api/v1/modules/core/permissions/service.py b/backend/api/v1/modules/core/permissions/service.py new file mode 100644 index 00000000..f82b906f --- /dev/null +++ b/backend/api/v1/modules/core/permissions/service.py @@ -0,0 +1,298 @@ +""" +Servicio de gestión de permisos multi-tenant. +Proporciona funciones para verificar y obtener permisos de usuarios por companye. +""" + +from datetime import datetime +from typing import Set, Optional, List +from sqlalchemy.orm import Session +from sqlalchemy import and_, or_ +from .models import ( + Permission, + CompanyRole, + RolePermission, + UserCompanyRole, + UserCompanyPermission, +) + + +class PermissionService: + """ + Servicio para gestionar permisos de usuarios en contextos multi-tenant. + Combina permisos de roles y permisos directos del usuario. + """ + + def __init__(self, db: Session): + self.db = db + + def get_user_permissions( + self, user_id: str, company_id: int, use_cache: bool = False + ) -> Set[str]: + """ + Obtiene todos los permisos de un usuario para un companye específico. + Combina permisos de roles asignados y permisos directos. + + Args: + user_id: ID del usuario + company_id: ID del companye/tenant + use_cache: Si True, intenta usar caché (útil en producción) + + Returns: + Set de códigos de permisos (ej: {"invoice.view", "invoice.edit"}) + """ + # Obtener permisos de roles asignados al usuario + role_permissions = self._get_permissions_from_roles(user_id, company_id) + + # Obtener permisos directos del usuario + direct_permissions = self._get_direct_permissions(user_id, company_id) + + # Combinar ambos conjuntos + # Los permisos directos revocados (is_granted=False) anulan los de roles + all_permissions = role_permissions.copy() + + for perm_code, is_granted in direct_permissions.items(): + if is_granted: + all_permissions.add(perm_code) + else: + # Permiso explícitamente revocado + all_permissions.discard(perm_code) + + return all_permissions + + def _get_permissions_from_roles(self, user_id: str, company_id: int) -> Set[str]: + """ + Obtiene permisos derivados de los roles del usuario en el companye. + + Realiza un JOIN eficiente para obtener todos los permisos de los roles activos. + """ + query = ( + self.db.query(Permission.code) + .join(RolePermission, RolePermission.permission_id == Permission.id) + .join(CompanyRole, CompanyRole.id == RolePermission.company_role_id) + .join(UserCompanyRole, UserCompanyRole.company_role_id == CompanyRole.id) + .filter( + and_( + UserCompanyRole.user_id == user_id, + UserCompanyRole.company_id == company_id, + UserCompanyRole.is_active == True, + CompanyRole.is_active == True, + Permission.is_active == True, + ) + ) + ) + + results = query.all() + return {perm_code for (perm_code,) in results} + + def _get_direct_permissions(self, user_id: str, company_id: int) -> dict: + """ + Obtiene permisos directos asignados al usuario. + + Returns: + Dict con código de permiso como key y is_granted como value + { + "invoice.delete": True, # Permiso concedido + "user.delete": False # Permiso revocado explícitamente + } + """ + now = datetime.utcnow() + + query = ( + self.db.query(Permission.code, UserCompanyPermission.is_granted) + .join( + UserCompanyPermission, + UserCompanyPermission.permission_id == Permission.id, + ) + .filter( + and_( + UserCompanyPermission.user_id == user_id, + UserCompanyPermission.company_id == company_id, + UserCompanyPermission.is_active == True, + Permission.is_active == True, + or_( + UserCompanyPermission.expires_at.is_(None), + UserCompanyPermission.expires_at > now, + ), + ) + ) + ) + + results = query.all() + return {perm_code: is_granted for perm_code, is_granted in results} + + def has_permission( + self, user_id: str, company_id: int, permission_code: str + ) -> bool: + """ + Verifica si un usuario tiene un permiso específico en un companye. + + Args: + user_id: ID del usuario + company_id: ID del companye/tenant + permission_code: Código del permiso (ej: "invoice.edit") + + Returns: + True si el usuario tiene el permiso, False en caso contrario + """ + permissions = self.get_user_permissions(user_id, company_id) + return permission_code in permissions + + def has_any_permission( + self, user_id: str, company_id: int, permission_codes: List[str] + ) -> bool: + """ + Verifica si el usuario tiene al menos uno de los permisos especificados. + """ + permissions = self.get_user_permissions(user_id, company_id) + return any(perm in permissions for perm in permission_codes) + + def has_all_permissions( + self, user_id: str, company_id: int, permission_codes: List[str] + ) -> bool: + """ + Verifica si el usuario tiene todos los permisos especificados. + """ + permissions = self.get_user_permissions(user_id, company_id) + return all(perm in permissions for perm in permission_codes) + + def get_user_roles(self, user_id: str, company_id: int) -> List[CompanyRole]: + """ + Obtiene los roles activos de un usuario en un companye. + """ + query = ( + self.db.query(CompanyRole) + .join(UserCompanyRole, UserCompanyRole.company_role_id == CompanyRole.id) + .filter( + and_( + UserCompanyRole.user_id == user_id, + UserCompanyRole.company_id == company_id, + UserCompanyRole.is_active == True, + CompanyRole.is_active == True, + ) + ) + ) + + return query.all() + + def assign_role_to_user( + self, + user_id: str, + company_id: int, + role_id: int, + assigned_by: Optional[str] = None, + ) -> UserCompanyRole: + """ + Asigna un rol a un usuario en un companye específico. + """ + # Verificar que el rol pertenece al companye + role = ( + self.db.query(CompanyRole) + .filter( + and_( + CompanyRole.id == role_id, + CompanyRole.company_id == company_id, + CompanyRole.is_active == True, + ) + ) + .first() + ) + + if not role: + raise ValueError(f"Role {role_id} not found for company {company_id}") + + # Verificar si ya existe la asignación + existing = ( + self.db.query(UserCompanyRole) + .filter( + and_( + UserCompanyRole.user_id == user_id, + UserCompanyRole.company_id == company_id, + UserCompanyRole.company_role_id == role_id, + ) + ) + .first() + ) + + if existing: + if not existing.is_active: + existing.is_active = True + existing.assigned_at = datetime.utcnow() + existing.assigned_by = assigned_by + self.db.commit() + return existing + return existing + + # Crear nueva asignación + user_role = UserCompanyRole( + user_id=user_id, + company_id=company_id, + company_role_id=role_id, + assigned_by=assigned_by, + ) + + self.db.add(user_role) + self.db.commit() + self.db.refresh(user_role) + + return user_role + + def grant_direct_permission( + self, + user_id: str, + company_id: int, + permission_code: str, + assigned_by: Optional[str] = None, + expires_at: Optional[datetime] = None, + ) -> UserCompanyPermission: + """ + Concede un permiso directo a un usuario en un companye. + """ + # Obtener el permiso por código + permission = ( + self.db.query(Permission) + .filter( + and_(Permission.code == permission_code, Permission.is_active == True) + ) + .first() + ) + + if not permission: + raise ValueError(f"Permission {permission_code} not found") + + # Verificar si ya existe + existing = ( + self.db.query(UserCompanyPermission) + .filter( + and_( + UserCompanyPermission.user_id == user_id, + UserCompanyPermission.company_id == company_id, + UserCompanyPermission.permission_id == permission.id, + ) + ) + .first() + ) + + if existing: + existing.is_granted = True + existing.is_active = True + existing.assigned_at = datetime.utcnow() + existing.assigned_by = assigned_by + existing.expires_at = expires_at + self.db.commit() + return existing + + # Crear nuevo permiso directo + user_permission = UserCompanyPermission( + user_id=user_id, + company_id=company_id, + permission_id=permission.id, + is_granted=True, + assigned_by=assigned_by, + expires_at=expires_at, + ) + + self.db.add(user_permission) + self.db.commit() + self.db.refresh(user_permission) + + return user_permission diff --git a/backend/api/v1/modules/core/permissions/test_permissions.py b/backend/api/v1/modules/core/permissions/test_permissions.py new file mode 100644 index 00000000..183e0dd2 --- /dev/null +++ b/backend/api/v1/modules/core/permissions/test_permissions.py @@ -0,0 +1,424 @@ +""" +Tests unitarios para el módulo de permisos multi-tenant. +Ejecutar con: pytest backend/api/v1/modules/core/permissions/test_permissions.py -v +""" + +import pytest +from datetime import datetime, timedelta +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker, Session +from core.database import Base +from api.v1.modules.core.permissions.models import ( + Permission, + ClientRole, + RolePermission, + UserClientRole, + UserClientPermission, +) +from api.v1.modules.core.permissions.service import PermissionService + + +# ============================================================================ +# FIXTURES +# ============================================================================ + + +@pytest.fixture +def db_session(): + """Crea una sesión de base de datos en memoria para testing.""" + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + Session = sessionmaker(bind=engine) + session = Session() + + yield session + + session.close() + + +@pytest.fixture +def sample_permissions(db_session: Session): + """Crea permisos de ejemplo.""" + permissions = [ + Permission( + code="invoice.view", + description="Ver facturas", + module="invoice", + action="view", + ), + Permission( + code="invoice.create", + description="Crear facturas", + module="invoice", + action="create", + ), + Permission( + code="invoice.edit", + description="Editar facturas", + module="invoice", + action="edit", + ), + Permission( + code="invoice.delete", + description="Eliminar facturas", + module="invoice", + action="delete", + ), + Permission( + code="user.view", description="Ver usuarios", module="user", action="view" + ), + ] + + for perm in permissions: + db_session.add(perm) + + db_session.commit() + return permissions + + +@pytest.fixture +def sample_role(db_session: Session, sample_permissions): + """Crea un rol de ejemplo con permisos.""" + role = ClientRole( + client_id=1, name="Administrador", code="admin", description="Acceso completo" + ) + db_session.add(role) + db_session.commit() + + # Asignar permisos al rol + for perm in sample_permissions[:3]: # invoice.view, create, edit + role_perm = RolePermission(client_role_id=role.id, permission_id=perm.id) + db_session.add(role_perm) + + db_session.commit() + return role + + +# ============================================================================ +# TESTS DE PermissionService +# ============================================================================ + + +class TestPermissionService: + """Tests para el servicio de permisos.""" + + def test_get_permissions_from_role(self, db_session, sample_role): + """Test: obtener permisos desde roles asignados.""" + # Asignar rol a usuario + user_role = UserClientRole( + user_id="user-123", client_id=1, client_role_id=sample_role.id + ) + db_session.add(user_role) + db_session.commit() + + # Verificar permisos + service = PermissionService(db_session) + permissions = service.get_user_permissions("user-123", 1) + + assert "invoice.view" in permissions + assert "invoice.create" in permissions + assert "invoice.edit" in permissions + assert "invoice.delete" not in permissions # No está en el rol + + def test_has_permission_single(self, db_session, sample_role): + """Test: verificar un permiso específico.""" + user_role = UserClientRole( + user_id="user-123", client_id=1, client_role_id=sample_role.id + ) + db_session.add(user_role) + db_session.commit() + + service = PermissionService(db_session) + + assert service.has_permission("user-123", 1, "invoice.view") is True + assert service.has_permission("user-123", 1, "invoice.delete") is False + + def test_has_all_permissions(self, db_session, sample_role): + """Test: verificar múltiples permisos (todos requeridos).""" + user_role = UserClientRole( + user_id="user-123", client_id=1, client_role_id=sample_role.id + ) + db_session.add(user_role) + db_session.commit() + + service = PermissionService(db_session) + + # Tiene todos + assert ( + service.has_all_permissions( + "user-123", 1, ["invoice.view", "invoice.create"] + ) + is True + ) + + # Falta uno + assert ( + service.has_all_permissions( + "user-123", 1, ["invoice.view", "invoice.delete"] + ) + is False + ) + + def test_has_any_permission(self, db_session, sample_role): + """Test: verificar múltiples permisos (al menos uno).""" + user_role = UserClientRole( + user_id="user-123", client_id=1, client_role_id=sample_role.id + ) + db_session.add(user_role) + db_session.commit() + + service = PermissionService(db_session) + + # Tiene al menos uno + assert ( + service.has_any_permission( + "user-123", 1, ["invoice.view", "invoice.delete"] + ) + is True + ) + + # No tiene ninguno + assert ( + service.has_any_permission("user-123", 1, ["invoice.delete", "user.view"]) + is False + ) + + def test_direct_permission_grant(self, db_session, sample_permissions): + """Test: conceder permiso directo a usuario.""" + service = PermissionService(db_session) + + # Conceder permiso directo + user_perm = service.grant_direct_permission( + user_id="user-123", client_id=1, permission_code="invoice.delete" + ) + + assert user_perm.is_granted is True + assert user_perm.is_active is True + + # Verificar que el usuario tiene el permiso + permissions = service.get_user_permissions("user-123", 1) + assert "invoice.delete" in permissions + + def test_direct_permission_revoke( + self, db_session, sample_role, sample_permissions + ): + """Test: revocar permiso que viene de un rol.""" + # Asignar rol con permisos + user_role = UserClientRole( + user_id="user-123", client_id=1, client_role_id=sample_role.id + ) + db_session.add(user_role) + db_session.commit() + + service = PermissionService(db_session) + + # Verificar que tiene el permiso del rol + assert service.has_permission("user-123", 1, "invoice.view") is True + + # Revocar explícitamente + invoice_view = ( + db_session.query(Permission) + .filter(Permission.code == "invoice.view") + .first() + ) + + user_perm = UserClientPermission( + user_id="user-123", + client_id=1, + permission_id=invoice_view.id, + is_granted=False, # Revocado + ) + db_session.add(user_perm) + db_session.commit() + + # Ahora NO debe tener el permiso + assert service.has_permission("user-123", 1, "invoice.view") is False + + def test_permission_expiration(self, db_session, sample_permissions): + """Test: permisos con fecha de expiración.""" + service = PermissionService(db_session) + + # Permiso expirado + expired_perm = service.grant_direct_permission( + user_id="user-123", + client_id=1, + permission_code="invoice.delete", + expires_at=datetime.utcnow() - timedelta(days=1), # Ayer + ) + + # No debe aparecer en los permisos + permissions = service.get_user_permissions("user-123", 1) + assert "invoice.delete" not in permissions + + # Permiso futuro + future_perm = service.grant_direct_permission( + user_id="user-456", + client_id=1, + permission_code="invoice.delete", + expires_at=datetime.utcnow() + timedelta(days=7), # En 7 días + ) + + # Debe aparecer + permissions = service.get_user_permissions("user-456", 1) + assert "invoice.delete" in permissions + + def test_assign_role_to_user(self, db_session, sample_role): + """Test: asignar un rol a un usuario.""" + service = PermissionService(db_session) + + user_role = service.assign_role_to_user( + user_id="user-789", + client_id=1, + role_id=sample_role.id, + assigned_by="admin-001", + ) + + assert user_role.user_id == "user-789" + assert user_role.client_id == 1 + assert user_role.client_role_id == sample_role.id + assert user_role.assigned_by == "admin-001" + assert user_role.is_active is True + + def test_assign_role_invalid_client(self, db_session, sample_role): + """Test: intentar asignar rol de otro cliente falla.""" + service = PermissionService(db_session) + + with pytest.raises(ValueError, match="not found for client"): + service.assign_role_to_user( + user_id="user-789", + client_id=999, # Cliente incorrecto + role_id=sample_role.id, + ) + + def test_multiple_roles_same_user(self, db_session, sample_permissions): + """Test: usuario con múltiples roles combina permisos.""" + # Crear dos roles diferentes + role1 = ClientRole(client_id=1, code="viewer", name="Viewer") + role2 = ClientRole(client_id=1, code="editor", name="Editor") + db_session.add_all([role1, role2]) + db_session.commit() + + # Role1: solo view + rp1 = RolePermission( + client_role_id=role1.id, + permission_id=sample_permissions[0].id, # invoice.view + ) + + # Role2: create y edit + rp2 = RolePermission( + client_role_id=role2.id, + permission_id=sample_permissions[1].id, # invoice.create + ) + rp3 = RolePermission( + client_role_id=role2.id, + permission_id=sample_permissions[2].id, # invoice.edit + ) + db_session.add_all([rp1, rp2, rp3]) + db_session.commit() + + # Asignar ambos roles al usuario + ur1 = UserClientRole(user_id="user-multi", client_id=1, client_role_id=role1.id) + ur2 = UserClientRole(user_id="user-multi", client_id=1, client_role_id=role2.id) + db_session.add_all([ur1, ur2]) + db_session.commit() + + # Verificar que tiene permisos de ambos roles + service = PermissionService(db_session) + permissions = service.get_user_permissions("user-multi", 1) + + assert "invoice.view" in permissions + assert "invoice.create" in permissions + assert "invoice.edit" in permissions + assert "invoice.delete" not in permissions + + def test_inactive_role_no_permissions(self, db_session, sample_role): + """Test: roles inactivos no otorgan permisos.""" + # Asignar rol activo + user_role = UserClientRole( + user_id="user-inactive", client_id=1, client_role_id=sample_role.id + ) + db_session.add(user_role) + db_session.commit() + + service = PermissionService(db_session) + + # Verificar que tiene permisos + assert service.has_permission("user-inactive", 1, "invoice.view") is True + + # Desactivar el rol + sample_role.is_active = False + db_session.commit() + + # Ahora no debe tener permisos + assert service.has_permission("user-inactive", 1, "invoice.view") is False + + def test_get_user_roles(self, db_session, sample_role): + """Test: obtener roles de un usuario.""" + user_role = UserClientRole( + user_id="user-roles", client_id=1, client_role_id=sample_role.id + ) + db_session.add(user_role) + db_session.commit() + + service = PermissionService(db_session) + roles = service.get_user_roles("user-roles", 1) + + assert len(roles) == 1 + assert roles[0].code == "admin" + assert roles[0].name == "Administrador" + + +# ============================================================================ +# TESTS DE MODELOS +# ============================================================================ + + +class TestModels: + """Tests para los modelos de base de datos.""" + + def test_permission_creation(self, db_session): + """Test: crear un permiso.""" + perm = Permission( + code="test.action", + description="Permiso de prueba", + module="test", + action="action", + ) + db_session.add(perm) + db_session.commit() + + assert perm.id is not None + assert perm.is_active is True + assert perm.created_at is not None + + def test_client_role_unique_constraint(self, db_session): + """Test: código de rol debe ser único por cliente.""" + role1 = ClientRole(client_id=1, code="admin", name="Admin 1") + db_session.add(role1) + db_session.commit() + + # Intentar crear otro con mismo código en mismo cliente + role2 = ClientRole(client_id=1, code="admin", name="Admin 2") + db_session.add(role2) + + with pytest.raises(Exception): # Violación de constraint + db_session.commit() + + def test_role_permission_relationship( + self, db_session, sample_role, sample_permissions + ): + """Test: relación entre roles y permisos.""" + # El rol ya tiene permisos del fixture + assert len(sample_role.role_permissions) > 0 + + # Verificar que los permisos están relacionados + permission_codes = [rp.permission.code for rp in sample_role.role_permissions] + assert "invoice.view" in permission_codes + + +# ============================================================================ +# EJECUTAR TESTS +# ============================================================================ + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/backend/api/v1/modules/public/reference_data/trailer_types/seed.py b/backend/api/v1/modules/public/reference_data/trailer_types/seed.py new file mode 100644 index 00000000..e172cf3e --- /dev/null +++ b/backend/api/v1/modules/public/reference_data/trailer_types/seed.py @@ -0,0 +1,45 @@ +seed = [ + ("20", "20 ft Sea container open top"), + ("2B", "20 ft Sea container - closed top"), + ("40", "40 ft Sea container - open top"), + ("4B", "40 ft Sea container - closed top"), + ("BI", "Beverage rack trailer"), + ("CB", "Gooseneck trailer"), + ("CH", "Chassis"), + ("CL", "Other lenght Sea container - closed top"), + ("CU", "Other length Sea container - open top"), + ("CZ", "Refrigerated Container"), + ("DD", "Double Drop Trailer"), + ("DT", "Drop Back Trailer"), + ("FR", "Flat rack trailer"), + ("FT", "Flatbed/platform trailer"), + ("HC", "Hopper Trailer (covered)"), + ("HE", "Horse trailer"), + ("HO", "Hopper Trailer (open)"), + ("HP", "Hopper Trailer (covered Pneumatic Discharged)"), + ("L1", "Tank trailer (liquids) not heated/not insulated"), + ("L2", "Tank trailer (liquids) heated/not insulated"), + ("L3", "Tank trailer (liquids) not heated/insulated"), + ("L4", "Tank trailer (liquids) heated/insulate"), + ("LP", "Logging, pipe or pole trailer"), + ("LT", "Livestock trailer"), + ("NC", "No Equipment"), + ("OE", "Other"), + ("RD", "Fixed Rack, Double Drop Trailer"), + ("RG", "Gondola closed"), + ("RO", "Gondola open"), + ("RS", "Fixed Rack, Single Drop Trailer"), + ("SD", "Single Drop Trailer"), + ("T1", "Tank trailer (gas) not heated/ not insulated"), + ("T2", "Tank trailer (gas) heated/not insulated"), + ("T3", "Tank trailer (gas) not heated/insulated"), + ("T4", "Tank trailer (gas) heated/insulate"), + ("T5", "Tank trailer (chemicals) not heated/not insulated"), + ("T6", "Tank trailer (chemicals) heated/not insulated"), + ("T7", "Tank trailer (chemicals) not heated/insulated"), + ("T8", "Tank trailer (chemicals) heated/insulate"), + ("TC", "Auto car/trailer"), + ("TK", "Tank trailer (food grade liquids)"), + ("TL", "Semi-truck trailer"), + ("TW", "Controlled temperature trailer"), +] \ No newline at end of file diff --git a/frontend/src/routes/dashboard/+page.svelte b/frontend/src/routes/dashboard/+page.svelte index 1c81c69d..1c28eec3 100644 --- a/frontend/src/routes/dashboard/+page.svelte +++ b/frontend/src/routes/dashboard/+page.svelte @@ -12,12 +12,10 @@ FileText, Users, Package, - TruckIcon, Clock, BarChart3, - PieChart, - TrendingUp, - AlertCircle, + Calendar, + CircleAlert, RefreshCw } from 'lucide-svelte'; import type { DashboardStats } from '$lib/api/dashboard/types'; @@ -28,6 +26,13 @@ let loading = $state(true); let error = $state(null); + let greeting = $derived(() => { + const hour = new Date().getHours(); + if (hour < 12) return 'Buenos días'; + if (hour < 18) return 'Buenas tardes'; + return 'Buenas noches'; + }); + async function loadDashboardData() { if (!companyStore.activeCompany) { error = 'No hay compañía activa seleccionada'; @@ -56,29 +61,39 @@
- -
-
-
-

Dashboard

-

+ {#snippet headerSection()} +

+
+

+ {greeting()}, equipo. +

+

{#if stats?.company_name} - {stats.company_name} - Sistema de gestión de comercio exterior + {stats.company_name} + Anexos 22/24/30 {:else} - Sistema de gestión de comercio exterior conforme a Anexos 24, 30 y 22 del SAT + Sistema de gestión de comercio exterior {/if}

- +
+ + +
-
+ {/snippet} + + {@render headerSection()} {#if error} - + Error {error} diff --git a/frontend/src/routes/dashboard/invoices/+page.svelte b/frontend/src/routes/dashboard/invoices/+page.svelte index 93f23c50..0c70436d 100644 --- a/frontend/src/routes/dashboard/invoices/+page.svelte +++ b/frontend/src/routes/dashboard/invoices/+page.svelte @@ -11,7 +11,7 @@ import type { PageData } from './$types'; import { browser } from '$app/environment'; import { companyStore } from '$lib/stores/company.svelte'; - import { Plus, Filter, Trash2, RefreshCw } from 'lucide-svelte'; + import { Plus, RefreshCw } from 'lucide-svelte'; // Los datos iniciales vienen del servidor let { data }: { data: PageData } = $props();