Spaces:
Paused
Paused
| """ | |
| Schedule models for script execution scheduling. | |
| """ | |
| import uuid | |
| from django.conf import settings | |
| from django.db import models | |
| from .script import Script | |
| class ScriptSchedule(models.Model): | |
| """ | |
| Represents a schedule configuration for a Script. | |
| Links to django-q2 Schedule objects for actual task scheduling. | |
| """ | |
| class RunMode(models.TextChoices): | |
| MANUAL = "manual", "Manual" | |
| INTERVAL = "interval", "Interval" | |
| DAILY = "daily", "Daily" | |
| WEEKLY = "weekly", "Weekly" | |
| MONTHLY = "monthly", "Monthly" | |
| class IntervalChoice(models.IntegerChoices): | |
| FIVE_MINUTES = 5, "Every 5 minutes" | |
| TEN_MINUTES = 10, "Every 10 minutes" | |
| FIFTEEN_MINUTES = 15, "Every 15 minutes" | |
| THIRTY_MINUTES = 30, "Every 30 minutes" | |
| ONE_HOUR = 60, "Every hour" | |
| TWO_HOURS = 120, "Every 2 hours" | |
| SIX_HOURS = 360, "Every 6 hours" | |
| TWELVE_HOURS = 720, "Every 12 hours" | |
| id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) | |
| # One-to-one relationship with Script | |
| script = models.OneToOneField( | |
| Script, | |
| on_delete=models.CASCADE, | |
| related_name="schedule", | |
| ) | |
| # Run mode selection | |
| run_mode = models.CharField( | |
| max_length=20, | |
| choices=RunMode.choices, | |
| default=RunMode.MANUAL, | |
| ) | |
| # Interval mode configuration | |
| interval_minutes = models.PositiveIntegerField( | |
| null=True, | |
| blank=True, | |
| choices=IntervalChoice.choices, | |
| help_text="Interval in minutes (for interval mode)", | |
| ) | |
| # Daily mode configuration - stored as JSON array of time strings | |
| daily_times = models.JSONField( | |
| default=list, | |
| blank=True, | |
| help_text='List of times in HH:MM format, e.g., ["09:00", "18:00"]', | |
| ) | |
| # Timezone for scheduled runs | |
| timezone = models.CharField( | |
| max_length=50, | |
| default="UTC", | |
| help_text="Timezone for scheduled runs (e.g., 'America/New_York')", | |
| ) | |
| # Weekly mode configuration | |
| weekly_days = models.JSONField( | |
| default=list, | |
| blank=True, | |
| help_text='Days of week [0-6] where 0=Monday, e.g., [0, 2, 4] for Mon/Wed/Fri', | |
| ) | |
| weekly_times = models.JSONField( | |
| default=list, | |
| blank=True, | |
| help_text='List of times in HH:MM format for weekly mode', | |
| ) | |
| # Monthly mode configuration | |
| monthly_days = models.JSONField( | |
| default=list, | |
| blank=True, | |
| help_text='Days of month [1-31], e.g., [1, 15] for 1st and 15th', | |
| ) | |
| monthly_times = models.JSONField( | |
| default=list, | |
| blank=True, | |
| help_text='List of times in HH:MM format for monthly mode', | |
| ) | |
| # Schedule state | |
| is_active = models.BooleanField( | |
| default=True, | |
| help_text="Whether this schedule is active (can be paused without deleting)", | |
| ) | |
| # Link to django-q2 Schedule objects (can be multiple for daily with multiple times) | |
| # Stored as JSON array of django-q2 Schedule IDs | |
| q_schedule_ids = models.JSONField( | |
| default=list, | |
| blank=True, | |
| help_text="django-q2 Schedule object IDs", | |
| ) | |
| # Tracking | |
| next_run = models.DateTimeField( | |
| null=True, | |
| blank=True, | |
| help_text="Cached next scheduled run time", | |
| ) | |
| last_scheduled_run = models.DateTimeField( | |
| null=True, | |
| blank=True, | |
| help_text="When the schedule last triggered a run", | |
| ) | |
| # Metadata | |
| created_at = models.DateTimeField(auto_now_add=True) | |
| updated_at = models.DateTimeField(auto_now=True) | |
| created_by = models.ForeignKey( | |
| settings.AUTH_USER_MODEL, | |
| on_delete=models.SET_NULL, | |
| null=True, | |
| blank=True, | |
| related_name="created_schedules", | |
| ) | |
| class Meta: | |
| db_table = "script_schedules" | |
| verbose_name = "script schedule" | |
| verbose_name_plural = "script schedules" | |
| def __str__(self): | |
| return f"Schedule for {self.script.name} ({self.run_mode})" | |
| def is_scheduled(self) -> bool: | |
| """Check if this script has an active schedule (not manual).""" | |
| return self.run_mode != self.RunMode.MANUAL and self.is_active | |
| DAY_NAMES = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] | |
| def schedule_display(self) -> str: | |
| """Human-readable schedule description.""" | |
| if self.run_mode == self.RunMode.MANUAL: | |
| return "Manual" | |
| elif self.run_mode == self.RunMode.INTERVAL: | |
| # Find the display label from choices | |
| for value, label in self.IntervalChoice.choices: | |
| if value == self.interval_minutes: | |
| return label | |
| return f"Every {self.interval_minutes} minutes" | |
| elif self.run_mode == self.RunMode.DAILY: | |
| times = ", ".join(self.daily_times) if self.daily_times else "No times set" | |
| return f"Daily at {times} ({self.timezone})" | |
| elif self.run_mode == self.RunMode.WEEKLY: | |
| days = ", ".join(self.DAY_NAMES[d] for d in sorted(self.weekly_days)) if self.weekly_days else "No days set" | |
| times = ", ".join(self.weekly_times) if self.weekly_times else "No times set" | |
| return f"Weekly on {days} at {times} ({self.timezone})" | |
| elif self.run_mode == self.RunMode.MONTHLY: | |
| days = ", ".join(str(d) for d in sorted(self.monthly_days)) if self.monthly_days else "No days set" | |
| times = ", ".join(self.monthly_times) if self.monthly_times else "No times set" | |
| return f"Monthly on day {days} at {times} ({self.timezone})" | |
| return "Unknown" | |
| class ScheduleHistory(models.Model): | |
| """ | |
| Tracks changes to script schedules for audit purposes. | |
| """ | |
| class ChangeType(models.TextChoices): | |
| CREATED = "created", "Created" | |
| UPDATED = "updated", "Updated" | |
| ENABLED = "enabled", "Enabled" | |
| DISABLED = "disabled", "Disabled" | |
| DELETED = "deleted", "Deleted" | |
| id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) | |
| schedule = models.ForeignKey( | |
| ScriptSchedule, | |
| on_delete=models.CASCADE, | |
| related_name="history", | |
| ) | |
| change_type = models.CharField( | |
| max_length=20, | |
| choices=ChangeType.choices, | |
| ) | |
| # Snapshot of schedule configuration at time of change | |
| previous_config = models.JSONField( | |
| null=True, | |
| blank=True, | |
| help_text="Previous schedule configuration", | |
| ) | |
| new_config = models.JSONField( | |
| null=True, | |
| blank=True, | |
| help_text="New schedule configuration", | |
| ) | |
| changed_by = models.ForeignKey( | |
| settings.AUTH_USER_MODEL, | |
| on_delete=models.SET_NULL, | |
| null=True, | |
| blank=True, | |
| ) | |
| created_at = models.DateTimeField(auto_now_add=True) | |
| class Meta: | |
| db_table = "schedule_history" | |
| verbose_name = "schedule history" | |
| verbose_name_plural = "schedule history entries" | |
| ordering = ["-created_at"] | |
| def __str__(self): | |
| return f"{self.change_type} - {self.schedule.script.name} at {self.created_at}" | |