feat: Add dedicated dashboard methods to service clients

Created typed, domain-specific methods in service clients instead of
using generic .get() calls with paths. This improves type safety,
discoverability, and maintainability.

Service Client Changes:
- ProcurementServiceClient:
  * get_pending_purchase_orders() - POs awaiting approval
  * get_critical_alerts() - Critical severity alerts
  * get_alerts_summary() - Alert counts by severity

- ProductionServiceClient:
  * get_todays_batches() - Today's production timeline
  * get_production_batches_by_status() - Filter by status

- InventoryServiceClient:
  * get_stock_status() - Dashboard stock metrics
  * get_sustainability_widget() - Sustainability data

Dashboard API Changes:
- Updated all endpoints to use new dedicated methods
- Cleaner, more maintainable code
- Better error handling and logging
- Fixed inventory data type handling (list vs dict)

Note: Alert endpoints return 404 - alert_processor service needs
endpoints: /alerts/summary and /alerts (filtered by severity).
This commit is contained in:
Claude
2025-11-07 22:12:21 +00:00
parent 9722cdb7f7
commit 6cd4ef0f56
4 changed files with 187 additions and 33 deletions

View File

@@ -573,3 +573,79 @@ class ProcurementServiceClient(BaseServiceClient):
logger.error("Error triggering price forecasting",
error=str(e), tenant_id=tenant_id)
return None
# ================================================================
# DASHBOARD METHODS
# ================================================================
async def get_pending_purchase_orders(
self,
tenant_id: str,
limit: int = 20
) -> Optional[Dict[str, Any]]:
"""
Get purchase orders pending approval for dashboard
Args:
tenant_id: Tenant ID
limit: Maximum number of POs to return
Returns:
Dict with {"items": [...], "total": n}
"""
try:
return await self.get(
"/procurement/purchase-orders",
tenant_id=tenant_id,
params={"status": "pending_approval", "limit": limit}
)
except Exception as e:
logger.error("Error fetching pending purchase orders", error=str(e), tenant_id=tenant_id)
return None
async def get_critical_alerts(
self,
tenant_id: str,
limit: int = 20
) -> Optional[Dict[str, Any]]:
"""
Get critical alerts for dashboard
Args:
tenant_id: Tenant ID
limit: Maximum number of alerts to return
Returns:
Dict with {"alerts": [...], "total": n}
"""
try:
return await self.get(
"/procurement/alert-processor/alerts",
tenant_id=tenant_id,
params={"severity": "critical", "resolved": False, "limit": limit}
)
except Exception as e:
logger.error("Error fetching critical alerts", error=str(e), tenant_id=tenant_id)
return None
async def get_alerts_summary(
self,
tenant_id: str
) -> Optional[Dict[str, Any]]:
"""
Get alerts summary for dashboard health status
Args:
tenant_id: Tenant ID
Returns:
Dict with counts by severity
"""
try:
return await self.get(
"/procurement/alert-processor/alerts/summary",
tenant_id=tenant_id
)
except Exception as e:
logger.error("Error fetching alerts summary", error=str(e), tenant_id=tenant_id)
return None