refactor: Extract alerts functionality to dedicated AlertsServiceClient

Moved alert-related methods from ProcurementServiceClient to a new
dedicated AlertsServiceClient for better separation of concerns.

Changes:
- Created shared/clients/alerts_client.py:
  * get_alerts_summary() - Alert counts by severity/status
  * get_critical_alerts() - Filtered list of urgent alerts
  * get_alerts_by_severity() - Filter by any severity level
  * get_alert_by_id() - Get specific alert details
  * Includes severity mapping (critical → urgent)

- Updated shared/clients/__init__.py:
  * Added AlertsServiceClient import/export
  * Added get_alerts_client() factory function

- Updated procurement_client.py:
  * Removed get_critical_alerts() method
  * Removed get_alerts_summary() method
  * Kept only procurement-specific methods

- Updated dashboard.py:
  * Import and initialize alerts_client
  * Use alerts_client for alert operations
  * Use procurement_client only for procurement operations

Benefits:
- Better separation of concerns
- Alerts logically grouped with alert_processor service
- Cleaner, more maintainable service client architecture
- Each client maps to its domain service
This commit is contained in:
Claude
2025-11-07 22:19:24 +00:00
parent eecd630d64
commit bc97fc0d1a
4 changed files with 193 additions and 56 deletions

View File

@@ -0,0 +1,173 @@
# shared/clients/alerts_client.py
"""
Alerts Service Client for Inter-Service Communication
Provides access to alert processor service from other services
"""
import structlog
from typing import Dict, Any, Optional, List
from uuid import UUID
from shared.clients.base_service_client import BaseServiceClient
from shared.config.base import BaseServiceSettings
logger = structlog.get_logger()
class AlertsServiceClient(BaseServiceClient):
"""Client for communicating with the Alert Processor Service"""
def __init__(self, config: BaseServiceSettings, calling_service_name: str = "unknown"):
super().__init__(calling_service_name, config)
def get_service_base_path(self) -> str:
return "/api/v1"
# ================================================================
# DASHBOARD METHODS
# ================================================================
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:
{
"total_count": int,
"active_count": int,
"critical_count": int, # Maps to "urgent" severity
"high_count": int,
"medium_count": int,
"low_count": int,
"resolved_count": int,
"acknowledged_count": int
}
"""
try:
# Gateway routes /tenants/{tenant_id}/alerts/... to alert_processor service
return await self.get(
"/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
async def get_critical_alerts(
self,
tenant_id: str,
limit: int = 20
) -> Optional[Dict[str, Any]]:
"""
Get critical/urgent alerts for dashboard
Note: "critical" in dashboard context maps to "urgent" severity in alert_processor
Args:
tenant_id: Tenant ID
limit: Maximum number of alerts to return
Returns:
Dict with:
{
"alerts": [...],
"total": int,
"limit": int,
"offset": int
}
"""
try:
# Gateway routes /tenants/{tenant_id}/alerts/... to alert_processor service
# "critical" in dashboard = "urgent" severity in alert_processor
return await self.get(
"/alerts",
tenant_id=tenant_id,
params={"severity": "urgent", "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_by_severity(
self,
tenant_id: str,
severity: str,
limit: int = 100,
resolved: Optional[bool] = None
) -> Optional[Dict[str, Any]]:
"""
Get alerts filtered by severity
Args:
tenant_id: Tenant ID
severity: Severity level (low, medium, high, urgent)
limit: Maximum number of alerts
resolved: Filter by resolved status (None = all, True = resolved only, False = unresolved only)
Returns:
Dict with alerts list and metadata
"""
try:
params = {"severity": severity, "limit": limit}
if resolved is not None:
params["resolved"] = resolved
return await self.get(
"/alerts",
tenant_id=tenant_id,
params=params
)
except Exception as e:
logger.error("Error fetching alerts by severity",
error=str(e), severity=severity, tenant_id=tenant_id)
return None
async def get_alert_by_id(
self,
tenant_id: str,
alert_id: str
) -> Optional[Dict[str, Any]]:
"""
Get a specific alert by ID
Args:
tenant_id: Tenant ID
alert_id: Alert UUID
Returns:
Dict with alert details
"""
try:
return await self.get(
f"/alerts/{alert_id}",
tenant_id=tenant_id
)
except Exception as e:
logger.error("Error fetching alert", error=str(e),
alert_id=alert_id, tenant_id=tenant_id)
return None
# ================================================================
# UTILITY METHODS
# ================================================================
async def health_check(self) -> bool:
"""Check if alerts service is healthy"""
try:
result = await self.get("../health") # Health endpoint is not tenant-scoped
return result is not None
except Exception as e:
logger.error("Alerts service health check failed", error=str(e))
return False
# Factory function for dependency injection
def create_alerts_client(config: BaseServiceSettings, calling_service_name: str = "unknown") -> AlertsServiceClient:
"""Create alerts service client instance"""
return AlertsServiceClient(config, calling_service_name)