# Imports
from django.db import models
# from django.utils.timezone import now
from django.utils.translation import gettext_lazy as _
from superdjango.db.audit.models import AddedByModel, ModifiedByModel
from .constants import DEFAULT_INTERVAL, FREQUENCIES, STATUS
# Exports
__all__ = (
"Activity",
"Job",
)
# Models
[docs]class Job(AddedByModel, ModifiedByModel):
"""A job to be scheduled."""
app_name = models.CharField(
_("app name"),
help_text=_("The app from which the job originates."),
max_length=128
)
at = models.TimeField(
_("at"),
blank=True,
help_text=_("The specific time at which the job will run."),
null=True
)
# Only displayed to administrators, and even then not editable.
callback = models.CharField(
_("callback"),
editable=False,
help_text=_("The callback that executes the job."),
max_length=256
)
description = models.TextField(
_("description"),
blank=True,
help_text=_("A brief description of the job."),
null=True
)
FREQUENCY_CHOICES = (
(FREQUENCIES.SECOND, _("Second")),
(FREQUENCIES.SECONDS, _("Seconds")),
(FREQUENCIES.MINUTE, _("Minute")),
(FREQUENCIES.MINUTES, _("Minutes")),
(FREQUENCIES.HOUR, _("Hour")),
(FREQUENCIES.HOURS, _("Hours")),
(FREQUENCIES.DAY, _("Day")),
(FREQUENCIES.DAYS, _("Days")),
(FREQUENCIES.WEEK, _("Week")),
(FREQUENCIES.WEEKS, _("Weeks")),
(FREQUENCIES.MONDAY, _("Monday")),
(FREQUENCIES.TUESDAY, _("Tuesday")),
(FREQUENCIES.WEDNESDAY, _("Wednesday")),
(FREQUENCIES.THURSDAY, _("Thursday")),
(FREQUENCIES.FRIDAY, _("Friday")),
(FREQUENCIES.SATURDAY, _("Saturday")),
(FREQUENCIES.SUNDAY, _("Sunday")),
)
frequency = models.CharField(
_("frequency"),
choices=FREQUENCY_CHOICES,
default=FREQUENCIES.MINUTE,
help_text=_("The frequency of the job."),
max_length=64
)
interval = models.PositiveSmallIntegerField(
_("interval"),
default=DEFAULT_INTERVAL,
help_text=_("The interval upon which the job runs.")
)
is_active = models.BooleanField(
_("active"),
default=True,
help_text=_("Indicates the job is enabled for execution.")
)
label = models.CharField(
_("label"),
help_text=_("The name or label of th job."),
max_length=128
)
class Meta:
get_latest_by = "added_dt"
ordering = ["-added_dt"]
unique_together = ["app_name", "label"]
verbose_name = _("Job")
verbose_name_plural = _("Jobs")
@property
def every(self):
"""Recombines interval and frequency.
:rtype: str
"""
return "%s %s" % (self.interval, self.frequency)
[docs]class Activity(models.Model):
"""The result of a scheduled job."""
added_dt = models.DateTimeField(
_("added date/time"),
auto_now_add=True,
help_text=_("Date and time the entry was recorded.")
)
# In case the job is deleted.
cached_job_label = models.CharField(
_("label"),
help_text=_("The name or label of th job."),
max_length=128
)
cached_success = models.BooleanField(
_("success"),
help_text=_("Indicates success for failure.")
)
elapsed_time = models.DurationField(
_("elapsed time"),
blank=True,
help_text=_("The time that passed from start to finish of job execution."),
null=True
)
end_dt = models.DateTimeField(
_("end date/time"),
help_text=_("Date and time the job finished."),
)
job = models.ForeignKey(
Job,
blank=True,
help_text=_("The job with which the activity is associated"),
null=True,
on_delete=models.SET_NULL,
related_name="activities",
verbose_name=_("job")
)
message = models.TextField(
_("message"),
blank=True,
help_text=_("The message from the activity that may be displayed to users."),
null=True
)
output = models.TextField(
_("output"),
blank=True,
help_text=_("The raw output of the command."),
null=True,
)
start_dt = models.DateTimeField(
_("start date/time"),
help_text=_("Date and time the job began.")
)
STATUS_CHOICES = (
(STATUS.CRITICAL, _("Critical")),
(STATUS.FAILURE, _("Failure")),
(STATUS.RETRY, _("Retry")),
(STATUS.SUCCESS, _("Success")),
(STATUS.UNKNOWN, _("Unknown")),
)
status = models.CharField(
choices=STATUS_CHOICES,
default=STATUS.UNKNOWN,
help_text=_("The status of the activity."),
max_length=64,
verbose_name=_("status")
)
class Meta:
get_latest_by = "added_dt"
ordering = ["-added_dt"]
verbose_name = _("Activity")
verbose_name_plural = _("Activities")
def __str__(self):
return self.cached_job_label
[docs] def get_css_color(self):
"""Get the CSS color style bass on the activity status.
:rtype: str
"""
if self.status == STATUS.CRITICAL:
return "bg-danger"
elif self.status == STATUS.FAILURE:
return "bg-danger"
elif self.status == STATUS.RETRY:
return "bg-warning"
elif self.status == STATUS.SUCCESS:
return "bg-success"
else:
return "bg-info"
[docs] def get_display_name(self):
"""Get a more verbose label for the activity.
:rtype: str
"""
return "%s %s" % (self.cached_job_label, self.added_dt)
[docs] def get_icon(self):
"""Get the icon (style) based on the activity status.
:rtype: str
"""
if self.status == STATUS.CRITICAL:
return "fas fa-skull text-white"
elif self.status == STATUS.FAILURE:
return "fas fa-exclamation-triangle text-white"
elif self.status == STATUS.RETRY:
return "fas fa-info-circle text-white"
elif self.status == STATUS.SUCCESS:
return "fas fa-check text-white"
else:
return "fas fa-question-circle text-white"
[docs] @classmethod
def log(cls, job_id, result):
"""Log activity for a given job.
:param job_id: The job ID (pk) of the job that was executed.
:type job_id: int
:param result: The result of the job.
:type result: superdjango.contrib.scheduler.library.Result
:rtype: Activity
"""
activity = cls(
elapsed_time=result.elapsed_time,
end_dt=result.end_dt,
job_id=job_id,
message=result.message,
output=result.output,
start_dt=result.start_dt,
status=result.status
)
activity.save()
return activity
[docs] def save(self, *args, **kwargs):
"""Send a message on critical status."""
if not self.elapsed_time:
self.elapsed_time = self.end_dt - self.start_dt
if self.job_id:
self.cached_job_label = self.job.label
if self.status == STATUS.SUCCESS:
self.cached_success = True
else:
self.cached_success = False
super().save(*args, **kwargs)