bakery-admin/bakery-ia
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Delete legacy alerts

Browse Source
This commit is contained in:
Urtzi Alfaro
2025-08-22 15:31:52 +02:00
parent c6dd6fd1de
commit 90100a66c6
40 changed files with 25 additions and 3308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
services/forecasting/app/api/forecasts.py
View File

@@ -12,7 +12,7 @@ import uuid
from app.services.forecasting_service import EnhancedForecastingService
from app.schemas.forecasts import (
ForecastRequest, ForecastResponse, BatchForecastRequest,
BatchForecastResponse, AlertResponse
BatchForecastResponse
)
from shared.auth.decorators import (
get_current_user_dep,
@@ -242,68 +242,6 @@ async def get_enhanced_tenant_forecasts(
)
@router.get("/tenants/{tenant_id}/forecasts/alerts")
@track_execution_time("enhanced_get_alerts_duration_seconds", "forecasting-service")
async def get_enhanced_forecast_alerts(
tenant_id: str = Path(..., description="Tenant ID"),
active_only: bool = Query(True, description="Return only active alerts"),
skip: int = Query(0, description="Number of records to skip"),
limit: int = Query(50, description="Number of records to return"),
request_obj: Request = None,
current_tenant: str = Depends(get_current_tenant_id_dep),
enhanced_forecasting_service: EnhancedForecastingService = Depends(get_enhanced_forecasting_service)
):
"""Get forecast alerts using enhanced repository pattern"""
metrics = get_metrics_collector(request_obj)
try:
# Enhanced tenant validation
if tenant_id != current_tenant:
if metrics:
metrics.increment_counter("enhanced_get_alerts_access_denied_total")
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access denied to tenant resources"
)
# Record metrics
if metrics:
metrics.increment_counter("enhanced_get_alerts_total")
# Get alerts using enhanced service
alerts = await enhanced_forecasting_service.get_tenant_alerts(
tenant_id=tenant_id,
active_only=active_only,
skip=skip,
limit=limit
)
if metrics:
metrics.increment_counter("enhanced_get_alerts_success_total")
return {
"tenant_id": tenant_id,
"alerts": alerts,
"total_returned": len(alerts),
"active_only": active_only,
"pagination": {
"skip": skip,
"limit": limit
},
"enhanced_features": True,
"repository_integration": True
}
except Exception as e:
if metrics:
metrics.increment_counter("enhanced_get_alerts_errors_total")
logger.error("Failed to get enhanced forecast alerts",
tenant_id=tenant_id,
error=str(e))
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to get forecast alerts"
)
@router.get("/tenants/{tenant_id}/forecasts/{forecast_id}")

4
services/forecasting/app/core/config.py
View File

@@ -51,9 +51,5 @@ class ForecastingSettings(BaseServiceSettings):
TEMPERATURE_THRESHOLD_HOT: float = float(os.getenv("TEMPERATURE_THRESHOLD_HOT", "30.0"))
RAIN_IMPACT_FACTOR: float = float(os.getenv("RAIN_IMPACT_FACTOR", "0.7"))
# Alert Thresholds
HIGH_DEMAND_THRESHOLD: float = float(os.getenv("HIGH_DEMAND_THRESHOLD", "1.5"))
LOW_DEMAND_THRESHOLD: float = float(os.getenv("LOW_DEMAND_THRESHOLD", "0.5"))
STOCKOUT_RISK_THRESHOLD: float = float(os.getenv("STOCKOUT_RISK_THRESHOLD", "0.9"))
settings = ForecastingSettings()

25
services/forecasting/app/models/forecasts.py
View File

@@ -86,29 +86,4 @@ class PredictionBatch(Base):
def __repr__(self):
return f"<PredictionBatch(id={self.id}, status={self.status})>"
class ForecastAlert(Base):
"""Alerts based on forecast results"""
__tablename__ = "forecast_alerts"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True)
forecast_id = Column(UUID(as_uuid=True), nullable=False)
# Alert information
alert_type = Column(String(50), nullable=False) # high_demand, low_demand, stockout_risk
severity = Column(String(20), default="medium") # low, medium, high, critical
message = Column(Text, nullable=False)
# Status
created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc))
acknowledged_at = Column(DateTime(timezone=True))
resolved_at = Column(DateTime(timezone=True))
is_active = Column(Boolean, default=True)
# Notification
notification_sent = Column(Boolean, default=False)
notification_method = Column(String(50)) # email, whatsapp, sms
def __repr__(self):
return f"<ForecastAlert(id={self.id}, type={self.alert_type})>"

2
services/forecasting/app/repositories/__init__.py
View File

@@ -6,7 +6,6 @@ Repository implementations for forecasting service
from .base import ForecastingBaseRepository
from .forecast_repository import ForecastRepository
from .prediction_batch_repository import PredictionBatchRepository
from .forecast_alert_repository import ForecastAlertRepository
from .performance_metric_repository import PerformanceMetricRepository
from .prediction_cache_repository import PredictionCacheRepository
@@ -14,7 +13,6 @@ __all__ = [
"ForecastingBaseRepository",
"ForecastRepository",
"PredictionBatchRepository",
"ForecastAlertRepository",
"PerformanceMetricRepository",
"PredictionCacheRepository"
]

375
services/forecasting/app/repositories/forecast_alert_repository.py
View File

@@ -1,375 +0,0 @@
"""
Forecast Alert Repository
Repository for forecast alert operations
"""
from typing import Optional, List, Dict, Any
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import text
from datetime import datetime, timedelta
import structlog
from .base import ForecastingBaseRepository
from app.models.forecasts import ForecastAlert
from shared.database.exceptions import DatabaseError, ValidationError
logger = structlog.get_logger()
class ForecastAlertRepository(ForecastingBaseRepository):
"""Repository for forecast alert operations"""
def __init__(self, session: AsyncSession, cache_ttl: Optional[int] = 300):
# Alerts change frequently, shorter cache time (5 minutes)
super().__init__(ForecastAlert, session, cache_ttl)
async def create_alert(self, alert_data: Dict[str, Any]) -> ForecastAlert:
"""Create a new forecast alert"""
try:
# Validate alert data
validation_result = self._validate_forecast_data(
alert_data,
["tenant_id", "forecast_id", "alert_type", "message"]
)
if not validation_result["is_valid"]:
raise ValidationError(f"Invalid alert data: {validation_result['errors']}")
# Set default values
if "severity" not in alert_data:
alert_data["severity"] = "medium"
if "is_active" not in alert_data:
alert_data["is_active"] = True
if "notification_sent" not in alert_data:
alert_data["notification_sent"] = False
alert = await self.create(alert_data)
logger.info("Forecast alert created",
alert_id=alert.id,
tenant_id=alert.tenant_id,
alert_type=alert.alert_type,
severity=alert.severity)
return alert
except ValidationError:
raise
except Exception as e:
logger.error("Failed to create forecast alert",
tenant_id=alert_data.get("tenant_id"),
error=str(e))
raise DatabaseError(f"Failed to create alert: {str(e)}")
async def get_active_alerts(
self,
tenant_id: str,
alert_type: str = None,
severity: str = None
) -> List[ForecastAlert]:
"""Get active alerts for a tenant"""
try:
filters = {
"tenant_id": tenant_id,
"is_active": True
}
if alert_type:
filters["alert_type"] = alert_type
if severity:
filters["severity"] = severity
return await self.get_multi(
filters=filters,
order_by="created_at",
order_desc=True
)
except Exception as e:
logger.error("Failed to get active alerts",
tenant_id=tenant_id,
error=str(e))
return []
async def acknowledge_alert(
self,
alert_id: str,
acknowledged_by: str = None
) -> Optional[ForecastAlert]:
"""Acknowledge an alert"""
try:
update_data = {
"acknowledged_at": datetime.utcnow()
}
if acknowledged_by:
# Store in message or create a new field if needed
current_alert = await self.get_by_id(alert_id)
if current_alert:
update_data["message"] = f"{current_alert.message} (Acknowledged by: {acknowledged_by})"
updated_alert = await self.update(alert_id, update_data)
logger.info("Alert acknowledged",
alert_id=alert_id,
acknowledged_by=acknowledged_by)
return updated_alert
except Exception as e:
logger.error("Failed to acknowledge alert",
alert_id=alert_id,
error=str(e))
raise DatabaseError(f"Failed to acknowledge alert: {str(e)}")
async def resolve_alert(
self,
alert_id: str,
resolved_by: str = None
) -> Optional[ForecastAlert]:
"""Resolve an alert"""
try:
update_data = {
"resolved_at": datetime.utcnow(),
"is_active": False
}
if resolved_by:
current_alert = await self.get_by_id(alert_id)
if current_alert:
update_data["message"] = f"{current_alert.message} (Resolved by: {resolved_by})"
updated_alert = await self.update(alert_id, update_data)
logger.info("Alert resolved",
alert_id=alert_id,
resolved_by=resolved_by)
return updated_alert
except Exception as e:
logger.error("Failed to resolve alert",
alert_id=alert_id,
error=str(e))
raise DatabaseError(f"Failed to resolve alert: {str(e)}")
async def mark_notification_sent(
self,
alert_id: str,
notification_method: str
) -> Optional[ForecastAlert]:
"""Mark alert notification as sent"""
try:
update_data = {
"notification_sent": True,
"notification_method": notification_method
}
updated_alert = await self.update(alert_id, update_data)
logger.debug("Alert notification marked as sent",
alert_id=alert_id,
method=notification_method)
return updated_alert
except Exception as e:
logger.error("Failed to mark notification as sent",
alert_id=alert_id,
error=str(e))
return None
async def get_unnotified_alerts(self, tenant_id: str = None) -> List[ForecastAlert]:
"""Get alerts that haven't been notified yet"""
try:
filters = {
"is_active": True,
"notification_sent": False
}
if tenant_id:
filters["tenant_id"] = tenant_id
return await self.get_multi(
filters=filters,
order_by="created_at",
order_desc=False # Oldest first for notification
)
except Exception as e:
logger.error("Failed to get unnotified alerts",
tenant_id=tenant_id,
error=str(e))
return []
async def get_alert_statistics(self, tenant_id: str) -> Dict[str, Any]:
"""Get alert statistics for a tenant"""
try:
# Get counts by type
type_query = text("""
SELECT alert_type, COUNT(*) as count
FROM forecast_alerts
WHERE tenant_id = :tenant_id
GROUP BY alert_type
ORDER BY count DESC
""")
result = await self.session.execute(type_query, {"tenant_id": tenant_id})
alerts_by_type = {row.alert_type: row.count for row in result.fetchall()}
# Get counts by severity
severity_query = text("""
SELECT severity, COUNT(*) as count
FROM forecast_alerts
WHERE tenant_id = :tenant_id
GROUP BY severity
ORDER BY count DESC
""")
severity_result = await self.session.execute(severity_query, {"tenant_id": tenant_id})
alerts_by_severity = {row.severity: row.count for row in severity_result.fetchall()}
# Get status counts
total_alerts = await self.count(filters={"tenant_id": tenant_id})
active_alerts = await self.count(filters={
"tenant_id": tenant_id,
"is_active": True
})
acknowledged_alerts = await self.count(filters={
"tenant_id": tenant_id,
"acknowledged_at": "IS NOT NULL" # This won't work with our current filters
})
# Get recent activity (alerts in last 7 days)
seven_days_ago = datetime.utcnow() - timedelta(days=7)
recent_alerts = len(await self.get_by_date_range(
tenant_id, seven_days_ago, datetime.utcnow(), limit=1000
))
# Calculate response metrics
response_query = text("""
SELECT
AVG(EXTRACT(EPOCH FROM (acknowledged_at - created_at))/60) as avg_acknowledgment_time_minutes,
AVG(EXTRACT(EPOCH FROM (resolved_at - created_at))/60) as avg_resolution_time_minutes,
COUNT(CASE WHEN acknowledged_at IS NOT NULL THEN 1 END) as acknowledged_count,
COUNT(CASE WHEN resolved_at IS NOT NULL THEN 1 END) as resolved_count
FROM forecast_alerts
WHERE tenant_id = :tenant_id
""")
response_result = await self.session.execute(response_query, {"tenant_id": tenant_id})
response_row = response_result.fetchone()
return {
"total_alerts": total_alerts,
"active_alerts": active_alerts,
"resolved_alerts": total_alerts - active_alerts,
"alerts_by_type": alerts_by_type,
"alerts_by_severity": alerts_by_severity,
"recent_alerts_7d": recent_alerts,
"response_metrics": {
"avg_acknowledgment_time_minutes": float(response_row.avg_acknowledgment_time_minutes or 0),
"avg_resolution_time_minutes": float(response_row.avg_resolution_time_minutes or 0),
"acknowledgment_rate": round((response_row.acknowledged_count / max(total_alerts, 1)) * 100, 2),
"resolution_rate": round((response_row.resolved_count / max(total_alerts, 1)) * 100, 2)
} if response_row else {
"avg_acknowledgment_time_minutes": 0.0,
"avg_resolution_time_minutes": 0.0,
"acknowledgment_rate": 0.0,
"resolution_rate": 0.0
}
}
except Exception as e:
logger.error("Failed to get alert statistics",
tenant_id=tenant_id,
error=str(e))
return {
"total_alerts": 0,
"active_alerts": 0,
"resolved_alerts": 0,
"alerts_by_type": {},
"alerts_by_severity": {},
"recent_alerts_7d": 0,
"response_metrics": {
"avg_acknowledgment_time_minutes": 0.0,
"avg_resolution_time_minutes": 0.0,
"acknowledgment_rate": 0.0,
"resolution_rate": 0.0
}
}
async def cleanup_old_alerts(self, days_old: int = 90) -> int:
"""Clean up old resolved alerts"""
try:
cutoff_date = datetime.utcnow() - timedelta(days=days_old)
query_text = """
DELETE FROM forecast_alerts
WHERE is_active = false
AND resolved_at IS NOT NULL
AND resolved_at < :cutoff_date
"""
result = await self.session.execute(text(query_text), {"cutoff_date": cutoff_date})
deleted_count = result.rowcount
logger.info("Cleaned up old forecast alerts",
deleted_count=deleted_count,
days_old=days_old)
return deleted_count
except Exception as e:
logger.error("Failed to cleanup old alerts",
error=str(e))
raise DatabaseError(f"Alert cleanup failed: {str(e)}")
async def bulk_resolve_alerts(
self,
tenant_id: str,
alert_type: str = None,
older_than_hours: int = 24
) -> int:
"""Bulk resolve old alerts"""
try:
cutoff_time = datetime.utcnow() - timedelta(hours=older_than_hours)
conditions = [
"tenant_id = :tenant_id",
"is_active = true",
"created_at < :cutoff_time"
]
params = {
"tenant_id": tenant_id,
"cutoff_time": cutoff_time
}
if alert_type:
conditions.append("alert_type = :alert_type")
params["alert_type"] = alert_type
query_text = f"""
UPDATE forecast_alerts
SET is_active = false, resolved_at = :resolved_at
WHERE {' AND '.join(conditions)}
"""
params["resolved_at"] = datetime.utcnow()
result = await self.session.execute(text(query_text), params)
resolved_count = result.rowcount
logger.info("Bulk resolved old alerts",
tenant_id=tenant_id,
alert_type=alert_type,
resolved_count=resolved_count,
older_than_hours=older_than_hours)
return resolved_count
except Exception as e:
logger.error("Failed to bulk resolve alerts",
tenant_id=tenant_id,
error=str(e))
raise DatabaseError(f"Bulk resolve failed: {str(e)}")

17
services/forecasting/app/schemas/forecasts.py
View File

@@ -14,11 +14,6 @@ class BusinessType(str, Enum):
INDIVIDUAL = "individual"
CENTRAL_WORKSHOP = "central_workshop"
class AlertType(str, Enum):
HIGH_DEMAND = "high_demand"
LOW_DEMAND = "low_demand"
STOCKOUT_RISK = "stockout_risk"
OVERPRODUCTION = "overproduction"
class ForecastRequest(BaseModel):
"""Request schema for generating forecasts"""
@@ -100,16 +95,4 @@ class BatchForecastResponse(BaseModel):
forecasts: Optional[List[ForecastResponse]]
error_message: Optional[str]
class AlertResponse(BaseModel):
"""Response schema for forecast alerts"""
id: str
tenant_id: str
forecast_id: str
alert_type: str
severity: str
message: str
is_active: bool
created_at: datetime
acknowledged_at: Optional[datetime]
notification_sent: bool

2
services/forecasting/app/services/__init__.py
View File

@@ -10,7 +10,6 @@ from .data_client import DataClient
from .messaging import (
publish_forecast_generated,
publish_batch_forecast_completed,
publish_forecast_alert,
ForecastingStatusPublisher
)
@@ -22,6 +21,5 @@ __all__ = [
"DataClient",
"publish_forecast_generated",
"publish_batch_forecast_completed",
"publish_forecast_alert",
"ForecastingStatusPublisher"
]

65
services/forecasting/app/services/forecasting_service.py
View File

@@ -18,7 +18,6 @@ from app.services.data_client import DataClient
from app.repositories import (
ForecastRepository,
PredictionBatchRepository,
ForecastAlertRepository,
PerformanceMetricRepository,
PredictionCacheRepository
)
@@ -36,7 +35,7 @@ logger = structlog.get_logger()
class EnhancedForecastingService:
"""
Enhanced forecasting service using repository pattern.
Handles forecast generation, batch processing, and alerting with proper data abstraction.
Handles forecast generation, batch processing with proper data abstraction.
"""
def __init__(self, database_manager=None):
@@ -55,7 +54,6 @@ class EnhancedForecastingService:
return {
'forecast': ForecastRepository(session),
'batch': PredictionBatchRepository(session),
'alert': ForecastAlertRepository(session),
'performance': PerformanceMetricRepository(session),
'cache': PredictionCacheRepository(session)
}
@@ -165,15 +163,6 @@ class EnhancedForecastingService:
logger.error("Failed to delete forecast", error=str(e))
return False
async def get_tenant_alerts(self, tenant_id: str, active_only: bool = True,
skip: int = 0, limit: int = 50) -> List[Dict]:
"""Get tenant alerts"""
try:
# Implementation would use repository pattern
return []
except Exception as e:
logger.error("Failed to get tenant alerts", error=str(e))
raise
async def get_tenant_forecast_statistics(self, tenant_id: str) -> Dict[str, Any]:
"""Get tenant forecast statistics"""
@@ -246,7 +235,7 @@ class EnhancedForecastingService:
request: ForecastRequest
) -> ForecastResponse:
"""
Generate forecast using repository pattern with caching and alerting.
Generate forecast using repository pattern with caching.
"""
start_time = datetime.utcnow()
@@ -339,8 +328,6 @@ class EnhancedForecastingService:
expires_in_hours=24
)
# Step 8: Check for alerts
await self._check_and_create_alerts(forecast, adjusted_prediction, repos)
logger.info("Enhanced forecast generated successfully",
forecast_id=forecast.id,
@@ -398,8 +385,6 @@ class EnhancedForecastingService:
# Get forecast summary
forecast_summary = await repos['forecast'].get_forecast_summary(tenant_id)
# Get alert statistics
alert_stats = await repos['alert'].get_alert_statistics(tenant_id)
# Get batch statistics
batch_stats = await repos['batch'].get_batch_statistics(tenant_id)
@@ -415,7 +400,6 @@ class EnhancedForecastingService:
return {
"tenant_id": tenant_id,
"forecast_analytics": forecast_summary,
"alert_analytics": alert_stats,
"batch_analytics": batch_stats,
"cache_performance": cache_stats,
"performance_trends": performance_trends,
@@ -469,51 +453,6 @@ class EnhancedForecastingService:
error=str(e))
raise DatabaseError(f"Failed to create batch: {str(e)}")
async def _check_and_create_alerts(self, forecast, prediction: Dict[str, Any], repos: Dict):
"""Check forecast results and create alerts if necessary"""
try:
alerts_to_create = []
# Check for high demand alert
if prediction['prediction'] > 100: # Threshold for high demand
alerts_to_create.append({
"tenant_id": str(forecast.tenant_id),
"forecast_id": str(forecast.id), # Convert UUID to string
"alert_type": "high_demand",
"severity": "high" if prediction['prediction'] > 200 else "medium",
"message": f"High demand predicted for inventory product {str(forecast.inventory_product_id)}: {prediction['prediction']:.1f} units"
})
# Check for low demand alert
elif prediction['prediction'] < 10: # Threshold for low demand
alerts_to_create.append({
"tenant_id": str(forecast.tenant_id),
"forecast_id": str(forecast.id), # Convert UUID to string
"alert_type": "low_demand",
"severity": "low",
"message": f"Low demand predicted for inventory product {str(forecast.inventory_product_id)}: {prediction['prediction']:.1f} units"
})
# Check for stockout risk (very low prediction with narrow confidence interval)
confidence_interval = prediction['upper_bound'] - prediction['lower_bound']
if prediction['prediction'] < 5 and confidence_interval < 10:
alerts_to_create.append({
"tenant_id": str(forecast.tenant_id),
"forecast_id": str(forecast.id), # Convert UUID to string
"alert_type": "stockout_risk",
"severity": "critical",
"message": f"Stockout risk for inventory product {str(forecast.inventory_product_id)}: predicted {prediction['prediction']:.1f} units with high confidence"
})
# Create alerts
for alert_data in alerts_to_create:
await repos['alert'].create_alert(alert_data)
except Exception as e:
logger.error("Failed to create alerts",
forecast_id=forecast.id,
error=str(e))
# Don't raise - alerts are not critical for forecast generation
def _create_forecast_response_from_cache(self, cache_entry) -> ForecastResponse:
"""Create forecast response from cached entry"""

19
services/forecasting/app/services/messaging.py
View File

@@ -72,12 +72,6 @@ async def publish_forecast_completed(data: Dict[str, Any]):
event = ForecastGeneratedEvent(service_name="forecasting_service", data=data, event_type="forecast.completed")
await rabbitmq_client.publish_forecast_event(event_type="completed", forecast_data=event.to_dict())
async def publish_alert_created(data: Dict[str, Any]):
"""Publish alert created event"""
# Assuming 'alert.created' is a type of forecast event, or define a new exchange/publisher method
if rabbitmq_client:
event = ForecastGeneratedEvent(service_name="forecasting_service", data=data, event_type="alert.created")
await rabbitmq_client.publish_forecast_event(event_type="alert.created", forecast_data=event.to_dict())
async def publish_batch_completed(data: Dict[str, Any]):
"""Publish batch forecast completed event"""
@@ -181,19 +175,6 @@ async def publish_batch_forecast_completed(data: dict) -> bool:
logger.error("Failed to publish batch forecast event", error=str(e))
return False
async def publish_forecast_alert(data: dict) -> bool:
"""Publish forecast alert event"""
try:
if rabbitmq_client:
await rabbitmq_client.publish_event(
exchange="forecasting_events",
routing_key="forecast.alert",
message=data
)
return True
except Exception as e:
logger.error("Failed to publish forecast alert event", error=str(e))
return False
# Publisher class for compatibility