85 lines
2.2 KiB
Python
85 lines
2.2 KiB
Python
"""
|
|
Base classes for monitoring commands.
|
|
All monitoring commands are inherited from BaseMonitoringCommand.
|
|
"""
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from datetime import date, datetime, timedelta
|
|
from typing import Any, Dict, List, Optional
|
|
from zoneinfo import ZoneInfo
|
|
import uuid
|
|
|
|
from core.models.enums import EventTypeChoices
|
|
|
|
|
|
@dataclass
|
|
class MonitoringResult:
|
|
"""Result from a monitoring command execution."""
|
|
command_name: str
|
|
execution_date: date
|
|
summary: Dict[str, Any]
|
|
event_type: EventTypeChoices
|
|
entity_id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
|
emails_sent: int = 0
|
|
details: List[Dict[str, Any]] = field(default_factory=list)
|
|
metadata: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
class BaseMonitoringCommand(ABC):
|
|
"""
|
|
Abstract base class for monitoring commands.
|
|
|
|
Subclasses must implement:
|
|
- name: Unique command identifier
|
|
- event_type: EventTypeChoices for audit logging
|
|
- execute(): Main command logic
|
|
"""
|
|
|
|
# Default timezone for all monitoring commands
|
|
DEFAULT_TIMEZONE = ZoneInfo('America/New_York')
|
|
|
|
@property
|
|
@abstractmethod
|
|
def name(self) -> str:
|
|
"""Unique identifier for this command."""
|
|
pass
|
|
|
|
@property
|
|
@abstractmethod
|
|
def event_type(self) -> EventTypeChoices:
|
|
"""Event type to publish for audit logging."""
|
|
pass
|
|
|
|
@property
|
|
def description(self) -> str:
|
|
"""Human-readable description of what this command does."""
|
|
return f"Monitoring command: {self.name}"
|
|
|
|
@abstractmethod
|
|
def execute(self, **kwargs) -> MonitoringResult:
|
|
"""
|
|
Execute the monitoring command.
|
|
|
|
Args:
|
|
**kwargs: Command-specific arguments
|
|
|
|
Returns:
|
|
MonitoringResult with execution summary
|
|
"""
|
|
pass
|
|
|
|
def get_previous_day(self, timezone=None) -> date:
|
|
"""
|
|
Get the previous day's date in the specified timezone.
|
|
|
|
Args:
|
|
timezone: pytz timezone (defaults to America/New_York)
|
|
|
|
Returns:
|
|
date object for previous day
|
|
"""
|
|
tz = timezone or self.DEFAULT_TIMEZONE
|
|
now = datetime.now(tz)
|
|
yesterday = now - timedelta(days=1)
|
|
return yesterday.date()
|