1# Imports 

2 

3from django.db import models 

4# from django.utils.timezone import now 

5from django.utils.translation import gettext_lazy as _ 

6from superdjango.db.audit.models import AddedByModel, ModifiedByModel 

7from .constants import DEFAULT_INTERVAL, FREQUENCIES, STATUS 

8 

9# Exports 

10 

11__all__ = ( 

12 "Activity", 

13 "Job", 

14) 

15 

16# Models 

17 

18 

19class Job(AddedByModel, ModifiedByModel): 

20 """A job to be scheduled.""" 

21 

22 app_name = models.CharField( 

23 _("app name"), 

24 help_text=_("The app from which the job originates."), 

25 max_length=128 

26 ) 

27 

28 at = models.TimeField( 

29 _("at"), 

30 blank=True, 

31 help_text=_("The specific time at which the job will run."), 

32 null=True 

33 ) 

34 

35 # Only displayed to administrators, and even then not editable. 

36 callback = models.CharField( 

37 _("callback"), 

38 editable=False, 

39 help_text=_("The callback that executes the job."), 

40 max_length=256 

41 ) 

42 

43 description = models.TextField( 

44 _("description"), 

45 blank=True, 

46 help_text=_("A brief description of the job."), 

47 null=True 

48 ) 

49 

50 FREQUENCY_CHOICES = ( 

51 (FREQUENCIES.SECOND, _("Second")), 

52 (FREQUENCIES.SECONDS, _("Seconds")), 

53 (FREQUENCIES.MINUTE, _("Minute")), 

54 (FREQUENCIES.MINUTES, _("Minutes")), 

55 (FREQUENCIES.HOUR, _("Hour")), 

56 (FREQUENCIES.HOURS, _("Hours")), 

57 (FREQUENCIES.DAY, _("Day")), 

58 (FREQUENCIES.DAYS, _("Days")), 

59 (FREQUENCIES.WEEK, _("Week")), 

60 (FREQUENCIES.WEEKS, _("Weeks")), 

61 (FREQUENCIES.MONDAY, _("Monday")), 

62 (FREQUENCIES.TUESDAY, _("Tuesday")), 

63 (FREQUENCIES.WEDNESDAY, _("Wednesday")), 

64 (FREQUENCIES.THURSDAY, _("Thursday")), 

65 (FREQUENCIES.FRIDAY, _("Friday")), 

66 (FREQUENCIES.SATURDAY, _("Saturday")), 

67 (FREQUENCIES.SUNDAY, _("Sunday")), 

68 ) 

69 frequency = models.CharField( 

70 _("frequency"), 

71 choices=FREQUENCY_CHOICES, 

72 default=FREQUENCIES.MINUTE, 

73 help_text=_("The frequency of the job."), 

74 max_length=64 

75 ) 

76 

77 interval = models.PositiveSmallIntegerField( 

78 _("interval"), 

79 default=DEFAULT_INTERVAL, 

80 help_text=_("The interval upon which the job runs.") 

81 ) 

82 

83 is_active = models.BooleanField( 

84 _("active"), 

85 default=True, 

86 help_text=_("Indicates the job is enabled for execution.") 

87 ) 

88 

89 label = models.CharField( 

90 _("label"), 

91 help_text=_("The name or label of th job."), 

92 max_length=128 

93 ) 

94 

95 class Meta: 

96 get_latest_by = "added_dt" 

97 ordering = ["-added_dt"] 

98 unique_together = ["app_name", "label"] 

99 verbose_name = _("Job") 

100 verbose_name_plural = _("Jobs") 

101 

102 @property 

103 def every(self): 

104 """Recombines interval and frequency. 

105 

106 :rtype: str 

107 

108 """ 

109 return "%s %s" % (self.interval, self.frequency) 

110 

111 

112class Activity(models.Model): 

113 """The result of a scheduled job.""" 

114 

115 added_dt = models.DateTimeField( 

116 _("added date/time"), 

117 auto_now_add=True, 

118 help_text=_("Date and time the entry was recorded.") 

119 ) 

120 

121 # In case the job is deleted. 

122 cached_job_label = models.CharField( 

123 _("label"), 

124 help_text=_("The name or label of th job."), 

125 max_length=128 

126 ) 

127 

128 cached_success = models.BooleanField( 

129 _("success"), 

130 help_text=_("Indicates success for failure.") 

131 ) 

132 

133 elapsed_time = models.DurationField( 

134 _("elapsed time"), 

135 blank=True, 

136 help_text=_("The time that passed from start to finish of job execution."), 

137 null=True 

138 ) 

139 

140 end_dt = models.DateTimeField( 

141 _("end date/time"), 

142 help_text=_("Date and time the job finished."), 

143 ) 

144 

145 job = models.ForeignKey( 

146 Job, 

147 blank=True, 

148 help_text=_("The job with which the activity is associated"), 

149 null=True, 

150 on_delete=models.SET_NULL, 

151 related_name="activities", 

152 verbose_name=_("job") 

153 ) 

154 

155 message = models.TextField( 

156 _("message"), 

157 blank=True, 

158 help_text=_("The message from the activity that may be displayed to users."), 

159 null=True 

160 ) 

161 

162 output = models.TextField( 

163 _("output"), 

164 blank=True, 

165 help_text=_("The raw output of the command."), 

166 null=True, 

167 

168 ) 

169 

170 start_dt = models.DateTimeField( 

171 _("start date/time"), 

172 help_text=_("Date and time the job began.") 

173 ) 

174 

175 STATUS_CHOICES = ( 

176 (STATUS.CRITICAL, _("Critical")), 

177 (STATUS.FAILURE, _("Failure")), 

178 (STATUS.RETRY, _("Retry")), 

179 (STATUS.SUCCESS, _("Success")), 

180 (STATUS.UNKNOWN, _("Unknown")), 

181 ) 

182 status = models.CharField( 

183 choices=STATUS_CHOICES, 

184 default=STATUS.UNKNOWN, 

185 help_text=_("The status of the activity."), 

186 max_length=64, 

187 verbose_name=_("status") 

188 ) 

189 

190 class Meta: 

191 get_latest_by = "added_dt" 

192 ordering = ["-added_dt"] 

193 verbose_name = _("Activity") 

194 verbose_name_plural = _("Activities") 

195 

196 def __str__(self): 

197 return self.cached_job_label 

198 

199 def get_css_color(self): 

200 """Get the CSS color style bass on the activity status. 

201 

202 :rtype: str 

203 

204 """ 

205 if self.status == STATUS.CRITICAL: 

206 return "bg-danger" 

207 elif self.status == STATUS.FAILURE: 

208 return "bg-danger" 

209 elif self.status == STATUS.RETRY: 

210 return "bg-warning" 

211 elif self.status == STATUS.SUCCESS: 

212 return "bg-success" 

213 else: 

214 return "bg-info" 

215 

216 def get_display_name(self): 

217 """Get a more verbose label for the activity. 

218 

219 :rtype: str 

220 

221 """ 

222 return "%s %s" % (self.cached_job_label, self.added_dt) 

223 

224 def get_icon(self): 

225 """Get the icon (style) based on the activity status. 

226 

227 :rtype: str 

228 

229 """ 

230 if self.status == STATUS.CRITICAL: 

231 return "fas fa-skull text-white" 

232 elif self.status == STATUS.FAILURE: 

233 return "fas fa-exclamation-triangle text-white" 

234 elif self.status == STATUS.RETRY: 

235 return "fas fa-info-circle text-white" 

236 elif self.status == STATUS.SUCCESS: 

237 return "fas fa-check text-white" 

238 else: 

239 return "fas fa-question-circle text-white" 

240 

241 @classmethod 

242 def log(cls, job_id, result): 

243 """Log activity for a given job. 

244 

245 :param job_id: The job ID (pk) of the job that was executed. 

246 :type job_id: int 

247 

248 :param result: The result of the job. 

249 :type result: superdjango.contrib.scheduler.library.Result 

250 

251 :rtype: Activity 

252 

253 """ 

254 activity = cls( 

255 elapsed_time=result.elapsed_time, 

256 end_dt=result.end_dt, 

257 job_id=job_id, 

258 message=result.message, 

259 output=result.output, 

260 start_dt=result.start_dt, 

261 status=result.status 

262 ) 

263 activity.save() 

264 

265 return activity 

266 

267 def save(self, *args, **kwargs): 

268 """Send a message on critical status.""" 

269 if not self.elapsed_time: 

270 self.elapsed_time = self.end_dt - self.start_dt 

271 

272 if self.job_id: 

273 self.cached_job_label = self.job.label 

274 

275 if self.status == STATUS.SUCCESS: 

276 self.cached_success = True 

277 else: 

278 self.cached_success = False 

279 

280 super().save(*args, **kwargs)