Coverage for backend \ app \ Inventario \ repositories \ inventarioRepository.py: 20.97%

1from app.Inventario.models.inventarioModel import Inventario 

2from app.Productos.models.productoModel import Producto 

3from sqlalchemy.orm import joinedload 

4 

5class InventarioRepository: 

6 def __init__(self, dbSession): 

7 self.dbSession = dbSession 

8 

9 def listarInventarios(self): 

10 return self.dbSession.query(Inventario).options(joinedload(Inventario.producto)).all() 

11 

12 def obtenerPorId(self, idInventario: int): 

13 return self.dbSession.query(Inventario).options(joinedload(Inventario.producto)).filter(Inventario.idInventario == idInventario).first() 

14 

15 def obtenerPorProducto(self, idProducto: int): 

16 return self.dbSession.query(Inventario).filter(Inventario.idProducto == idProducto).first() 

17 

18 def crearInventario(self, inventario): 

19 # validar producto existe y está activo 

20 producto = self.dbSession.query(Producto).filter(Producto.idProducto == inventario.idProducto).first() 

21 missing = [] 

22 if not producto: 

23 missing.append("producto") 

24 else: 

25 if not getattr(producto, "activoProducto", True): 

26 missing.append("producto_inactiva") 

27 if missing: 

28 return {"error": missing} 

29 datos = inventario.model_dump() 

30 # garantizar que cantidadMinima tenga valor si no se envía 

31 if datos.get("cantidadMinima") is None: 

32 datos["cantidadMinima"] = 0 

33 nuevo = Inventario(**datos) 

34 self.dbSession.add(nuevo) 

35 self.dbSession.commit() 

36 self.dbSession.refresh(nuevo) 

37 return nuevo 

38 

39 def modificarInventario(self, idInventario: int, inventarioActualizar): 

40 inventario = self.obtenerPorId(idInventario) 

41 if not inventario: 

42 return None 

43 datos = inventarioActualizar.model_dump(exclude_unset=True) 

44 camposValidos = ["cantidadDisponible","cantidadMinima","activoInventario"] 

45 for campo, valor in datos.items(): 

46 if campo in camposValidos: 

47 setattr(inventario, campo, valor) 

48 self.dbSession.commit() 

49 self.dbSession.refresh(inventario) 

50 return inventario 

51 

52 def deshabilitarInventario(self, idInventario: int): 

53 inventario = self.obtenerPorId(idInventario) 

54 if not inventario: 

55 return None 

56 # Solo permitir deshabilitar si el producto asociado está deshabilitado 

57 producto = self.dbSession.query(Producto).filter(Producto.idProducto == inventario.idProducto).first() 

58 if producto and getattr(producto, "activoProducto", True): 

59 return False 

60 inventario.activoInventario = False 

61 self.dbSession.commit() 

62 self.dbSession.refresh(inventario) 

63 return inventario 

64 

65 def crearInventariosIniciales(self): 

66 # Crear un inventario por cada producto existente si no tiene inventario 

67 productos = self.dbSession.query(Producto).all() 

68 for p in productos: 

69 existe = self.dbSession.query(Inventario).filter(Inventario.idProducto == p.idProducto).first() 

70 if not existe: 

71 nuevo = Inventario(idProducto=p.idProducto, cantidadDisponible=0, cantidadMinima=0, activoInventario=True) 

72 self.dbSession.add(nuevo) 

73 print("Inventarios por defecto creados...!") 

74 self.dbSession.commit() 

75 self.dbSession.close()