224 lines
7.3 KiB
Python
224 lines
7.3 KiB
Python
# core/base_repository.py
|
|
from typing import Generic, TypeVar, Type, Optional, List, Dict, Any
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session, Query
|
|
from sqlalchemy import and_, desc
|
|
|
|
from database import Base
|
|
|
|
ModelType = TypeVar("ModelType", bound= Base)
|
|
|
|
|
|
class BaseRepository(Generic[ModelType]):
|
|
"""
|
|
Repositorio base con:
|
|
- CRUD completo
|
|
- Soft delete integrado
|
|
- Trazabilidad (created_by, updated_by, deleted_by)
|
|
- Métodos para root, admin, admin_licencias
|
|
"""
|
|
|
|
def __init__(self, db: Session, model: Type[ModelType]):
|
|
self.db = db
|
|
self.model = model
|
|
|
|
# ============================================
|
|
# QUERY BASE
|
|
# ============================================
|
|
|
|
def _base_query(self, include_deleted: bool = False) -> Query:
|
|
"""
|
|
Query base con filtro de soft delete
|
|
include_deleted=True: incluye registros eliminados
|
|
"""
|
|
query = self.db.query(self.model)
|
|
if not include_deleted:
|
|
query = query.filter(
|
|
self.model.deleted_at.is_(None),
|
|
self.model.is_active == True
|
|
)
|
|
return query
|
|
|
|
# ============================================
|
|
# CRUD BÁSICO
|
|
# ============================================
|
|
|
|
def get_by_id(self, id: int, include_deleted: bool = False) -> Optional[ModelType]:
|
|
"""Obtiene por ID"""
|
|
return self._base_query(include_deleted).filter(self.model.id == id).first()
|
|
|
|
def get_all(
|
|
self,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
filters: Dict[str, Any] = None,
|
|
order_by: str = None,
|
|
descending: bool = False,
|
|
include_deleted: bool = False
|
|
) -> List[ModelType]:
|
|
"""Lista con paginación y filtros"""
|
|
query = self._base_query(include_deleted)
|
|
|
|
# Aplicar filtros
|
|
if filters:
|
|
for key, value in filters.items():
|
|
if hasattr(self.model, key) and value is not None:
|
|
query = query.filter(getattr(self.model, key) == value)
|
|
|
|
# Ordenamiento
|
|
if order_by and hasattr(self.model, order_by):
|
|
order_col = getattr(self.model, order_by)
|
|
query = query.order_by(desc(order_col) if descending else order_col)
|
|
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def count(self, filters: Dict[str, Any] = None, include_deleted: bool = False) -> int:
|
|
"""Cuenta registros"""
|
|
query = self._base_query(include_deleted)
|
|
if filters:
|
|
for key, value in filters.items():
|
|
if hasattr(self.model, key) and value is not None:
|
|
query = query.filter(getattr(self.model, key) == value)
|
|
return query.count()
|
|
|
|
def create(self, data: Dict[str, Any], created_by: int = None) -> ModelType:
|
|
"""Crea un nuevo registro"""
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
# Eliminar campos que deben ser automáticos
|
|
data.pop("created_at", None)
|
|
data.pop("updated_at", None)
|
|
data.pop("deleted_at", None)
|
|
data.pop("last_login", None)
|
|
data.pop("id", None)
|
|
|
|
if created_by:
|
|
data["created_by"] = created_by
|
|
|
|
logger.info(f"CREANDO ENTIDAD {self.model.__name__} CON DATOS: {data}")
|
|
|
|
entity = self.model(**data)
|
|
self.db.add(entity)
|
|
self.db.commit()
|
|
self.db.refresh(entity)
|
|
return entity
|
|
|
|
except Exception as e:
|
|
logger.error(f"ERROR EN CREATE: {str(e)}")
|
|
self.db.rollback()
|
|
raise e
|
|
|
|
def update(
|
|
self,
|
|
id: int,
|
|
data: Dict[str, Any],
|
|
updated_by: int = None,
|
|
include_deleted: bool = False
|
|
) -> Optional[ModelType]:
|
|
"""Actualiza un registro"""
|
|
entity = self.get_by_id(id, include_deleted)
|
|
if not entity:
|
|
return None
|
|
|
|
# No permitir actualizar si está eliminado
|
|
if entity.deleted_at is not None and not include_deleted:
|
|
return None
|
|
|
|
# Actualizar campos
|
|
for key, value in data.items():
|
|
if hasattr(entity, key) and value is not None:
|
|
setattr(entity, key, value)
|
|
|
|
if updated_by:
|
|
entity.updated_by = updated_by
|
|
|
|
self.db.commit()
|
|
self.db.refresh(entity)
|
|
return entity
|
|
|
|
def soft_delete(self, id: int, deleted_by: int = None) -> bool:
|
|
"""Soft delete: marca is_active=False y deleted_at"""
|
|
entity = self.get_by_id(id)
|
|
if not entity:
|
|
return False
|
|
|
|
if entity.deleted_at is not None:
|
|
return False
|
|
|
|
entity.is_active = False
|
|
entity.deleted_at = datetime.utcnow()
|
|
if deleted_by:
|
|
entity.deleted_by = deleted_by
|
|
|
|
self.db.commit()
|
|
return True
|
|
|
|
def hard_delete(self, id: int) -> bool:
|
|
"""Eliminación física"""
|
|
entity = self.get_by_id(id, include_deleted=True)
|
|
if not entity:
|
|
return False
|
|
|
|
self.db.delete(entity)
|
|
self.db.commit()
|
|
return True
|
|
|
|
def restore(self, id: int, updated_by: int = None) -> Optional[ModelType]:
|
|
"""Restaura un registro soft-deleted"""
|
|
entity = self.get_by_id(id, include_deleted=True)
|
|
if not entity:
|
|
return None
|
|
|
|
if entity.deleted_at is None:
|
|
return entity
|
|
|
|
entity.is_active = True
|
|
entity.deleted_at = None
|
|
if updated_by:
|
|
entity.updated_by = updated_by
|
|
|
|
self.db.commit()
|
|
self.db.refresh(entity)
|
|
return entity
|
|
|
|
# ============================================
|
|
# MÉTODOS PARA DIFERENCIACIÓN DE USUARIOS
|
|
# ============================================
|
|
# Estos métodos serán sobrescritos en repositorios específicos
|
|
# si el modelo tiene los campos correspondientes
|
|
|
|
def get_by_user_type(self, user_type: str) -> List[ModelType]:
|
|
"""Obtiene por tipo de usuario (root, admin, admin_licencias, user)"""
|
|
if hasattr(self.model, "tipo_usuario"):
|
|
return self._base_query().filter(self.model.tipo_usuario == user_type).all()
|
|
return []
|
|
|
|
def get_root(self) -> List[ModelType]:
|
|
"""Obtiene usuarios tipo root"""
|
|
return self.get_by_user_type("root")
|
|
|
|
def get_admin(self) -> List[ModelType]:
|
|
"""Obtiene usuarios tipo admin"""
|
|
return self.get_by_user_type("admin")
|
|
|
|
def get_admin_licencias(self) -> List[ModelType]:
|
|
"""Obtiene usuarios tipo admin_licencias"""
|
|
return self.get_by_user_type("admin_licencias")
|
|
|
|
#===============================================
|
|
# el tipo de usuario "PRIVILEGED" un esta por definir que clase de permisos tendras
|
|
#=================================
|
|
|
|
def get_by_role(self, role: str) -> List[ModelType]:
|
|
"""Obtiene por rol operativo"""
|
|
if hasattr(self.model, "rol_operativo"):
|
|
return self._base_query().filter(self.model.rol_operativo == role).all()
|
|
return []
|
|
|
|
def get_by_enterprise(self, enterprise_id: int) -> List[ModelType]:
|
|
"""Obtiene por empresa"""
|
|
if hasattr(self.model, "enterprise_id"):
|
|
return self._base_query().filter(self.model.enterprise_id == enterprise_id).all()
|
|
return [] |