Documentation Index
Fetch the complete documentation index at: https://mintlify.com/ElthonJohan/Sistema-MRP/llms.txt
Use this file to discover all available pages before exploring further.
The dashboard exposes four KPI cards that give client users an immediate read on the health of their logistics operation. Each metric is computed by get_kpis() in services/dashboard_service.py and scoped to the authenticated client’s warehouses via an owner_id filter.
All KPIs are scoped per client. Every query in get_kpis() (and the supporting functions) first resolves the set of Warehouse.id values where Warehouse.owner_id == owner_id, then filters all further queries to that set. A superadmin viewing a client via impersonation uses the impersonated user’s owner_id, so the numbers are always client-specific.
Stock Total
What it shows: The total number of units currently on hand across all active inventory records for the client.
Formula:
SUM(inventory.stock) WHERE inventory.is_active = TRUE AND inventory.warehouse_id IN (client warehouses)
inv_q = db.query(Inventory).filter(Inventory.is_active == True)
if owner_id is not None:
wh_filter_ids = [
r[0] for r in db.query(Warehouse.id).filter(Warehouse.owner_id == owner_id).all()
]
inv_q = inv_q.filter(Inventory.warehouse_id.in_(wh_filter_ids))
total_stock = sum(s[0] for s in inv_q.with_entities(Inventory.stock).all())
"principal" and "obra"). The value includes both free stock and reserved stock — it reflects inventory.stock, not inventory.stock - inventory.reserved.
Materiales Críticos
What it shows: The count of inventory records in the client’s principal warehouse where stock has fallen to 5 units or fewer.
Formula:
COUNT(inventory) WHERE warehouse.type = 'principal'
AND inventory.stock <= 5
AND inventory.is_active = TRUE
AND inventory.warehouse_id IN (client warehouses)
crit_q = db.query(Inventory).join(Inventory.warehouse).filter(
Warehouse.type == "principal",
Inventory.stock <= 5,
Inventory.is_active == True,
)
if wh_filter_ids is not None:
crit_q = crit_q.filter(Inventory.warehouse_id.in_(wh_filter_ids))
critical = crit_q.count()
critical > 0, the KPI card displays an amber “Atención” badge. When critical == 0, it shows a neutral “Sin alertas” badge. The threshold of 5 is hard-coded in the query.
Only the principal warehouse is checked for critical stock. Work-site ("obra") warehouses are excluded from this count because they receive stock via dispatches rather than direct restocking.
Pendientes
What it shows: The count of requirements that have not yet been fully dispatched — those with status pending or partial.
Formula:
COUNT(requirement) WHERE requirement.status IN ('pending', 'partial')
AND requirement.warehouse_id_obra IN (client warehouses)
req_q = db.query(Requirement).filter(Requirement.status.in_(["pending", "partial"]))
if wh_filter_ids is not None:
req_q = req_q.filter(Requirement.warehouse_id_obra.in_(wh_filter_ids))
pending_req = req_q.count()
pending when created. It moves to partial when some — but not all — of its items have been dispatched. It reaches fulfilled only when every item is dispatched. Both pending and partial requirements are counted here because they still represent unfulfilled demand on the system.
Despachos
What it shows: The all-time total count of outbound (OUT) movement records for the client. Despite the internal dictionary key being dispatch_today, this is not filtered to the current day.
Formula:
COUNT(movement) WHERE movement.movement_type = 'OUT'
AND movement.warehouse_id IN (client warehouses)
mov_q = db.query(Movement).filter(Movement.movement_type == "OUT")
if wh_filter_ids is not None:
mov_q = mov_q.filter(Movement.warehouse_id.in_(wh_filter_ids))
dispatch_today = mov_q.count()
dispatch_service.create_dispatch() writes one Movement(type="OUT") record per dispatched item. The KPI aggregates all such records ever written for the client’s warehouses.
Full get_kpis function
def get_kpis(db: Session, owner_id: int = None):
"""KPIs generales. Si owner_id se pasa, filtra solo los almacenes del cliente."""
inv_q = db.query(Inventory).filter(Inventory.is_active == True)
wh_filter_ids = None
if owner_id is not None:
wh_filter_ids = [
r[0] for r in db.query(Warehouse.id).filter(Warehouse.owner_id == owner_id).all()
]
inv_q = inv_q.filter(Inventory.warehouse_id.in_(wh_filter_ids))
total_stock = sum(s[0] for s in inv_q.with_entities(Inventory.stock).all())
crit_q = db.query(Inventory).join(Inventory.warehouse).filter(
Warehouse.type == "principal",
Inventory.stock <= 5,
Inventory.is_active == True,
)
if wh_filter_ids is not None:
crit_q = crit_q.filter(Inventory.warehouse_id.in_(wh_filter_ids))
critical = crit_q.count()
req_q = db.query(Requirement).filter(Requirement.status.in_(["pending", "partial"]))
if wh_filter_ids is not None:
req_q = req_q.filter(Requirement.warehouse_id_obra.in_(wh_filter_ids))
pending_req = req_q.count()
mov_q = db.query(Movement).filter(Movement.movement_type == "OUT")
if wh_filter_ids is not None:
mov_q = mov_q.filter(Movement.warehouse_id.in_(wh_filter_ids))
dispatch_today = mov_q.count()
return {
"total_stock": total_stock,
"critical": critical,
"pending_req": pending_req,
"dispatch_today": dispatch_today,
}
Supporting chart functions
get_movements_last_7_days
Groups Movement records by calendar day over the last 7 days, returning separate IN and OUT counts for each day. Used to render the trend line chart on the dashboard.
def get_movements_last_7_days(db: Session, owner_id: int = None):
today = date.today()
days = [(today - timedelta(days=i)) for i in range(6, -1, -1)]
wh_ids = None
if owner_id is not None:
wh_ids = [
r[0] for r in db.query(Warehouse.id).filter(Warehouse.owner_id == owner_id).all()
]
result = []
for day in days:
start = datetime(day.year, day.month, day.day, 0, 0, 0)
end = datetime(day.year, day.month, day.day, 23, 59, 59)
def _count(mtype):
q = db.query(Movement).filter(
Movement.movement_type == mtype,
Movement.timestamp >= start,
Movement.timestamp <= end,
)
if wh_ids is not None:
q = q.filter(Movement.warehouse_id.in_(wh_ids))
return q.count()
result.append({
"Día": day.strftime("%d/%m"),
"Entradas": _count("IN"),
"Salidas": _count("OUT"),
})
return result
"DD/MM"), the count of IN movements, and the count of OUT movements for that day. Days with zero activity are included, producing a continuous 7-point series.
get_top_materials_by_movement
Returns the top 5 materials by total movement count (IN + OUT combined), used to render the horizontal bar chart.
def get_top_materials_by_movement(db: Session, limit: int = 5, owner_id: int = None):
q = db.query(Movement.material_id, func.count(Movement.id).label("total"))
if owner_id is not None:
wh_ids = [
r[0] for r in db.query(Warehouse.id).filter(Warehouse.owner_id == owner_id).all()
]
q = q.filter(Movement.warehouse_id.in_(wh_ids))
rows = q.group_by(Movement.material_id).order_by(func.count(Movement.id).desc()).limit(limit).all()
from models.material import Material
result = []
for r in rows:
mat = db.query(Material).filter(Material.id == r.material_id).first()
if mat:
result.append({"Material": mat.name, "Movimientos": r.total})
return result
Movement rows by material_id, counts them, orders descending, and resolves each material_id to its display name via a secondary Material lookup. The limit parameter defaults to 5 but can be overridden.