Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _info_suffix(self):
if self.is_periodic():
return ' ({s})'.format(s=colored('pink', self.schedule))
elif self.is_triggered():
return ' (on changed {t})'.format(t=colored('pink', self.table))
else:
return ''
def _info_suffix(self):
if self.is_periodic():
return ' ({s})'.format(s=colored('pink', self.schedule))
elif self.is_triggered():
return ' (on changed {t})'.format(t=colored('pink', self.table))
else:
return ''
def summary_info(self):
if self.summary is None:
return ''
if self.type == 'now':
return 'C:{c} S:{s} F:{f} L:{l} Q:{q} R:{r}'.format(
c=colored('green', self.summary['Complete']),
s=colored('blue', self.summary['Starting']),
f=colored('red', self.summary['Failed']),
l=colored('orange', self.summary['Lost']), # noqa: E741
q=colored('pink', self.summary['Queued']),
r=colored('yellow', self.summary['Running'])
)
else:
# For 'periodic', 'trigger', the summary is empty;
# it would be interesting to show the summary of the
# dispatch children jobs as part of the corresponding
# execution status.
return 'P:{p} R: {r} D:{d}'.format(
p=colored(_status_color('pending'), self.children['Pending']),
r=colored(_status_color('running'), self.children['Running']),
d=colored(_status_color('dead'), self.children['Dead'])
)
return 'C:{c} S:{s} F:{f} L:{l} Q:{q} R:{r}'.format(
c=colored('green', self.summary['Complete']),
s=colored('blue', self.summary['Starting']),
f=colored('red', self.summary['Failed']),
l=colored('orange', self.summary['Lost']), # noqa: E741
q=colored('pink', self.summary['Queued']),
r=colored('yellow', self.summary['Running'])
)
else:
# For 'periodic', 'trigger', the summary is empty;
# it would be interesting to show the summary of the
# dispatch children jobs as part of the corresponding
# execution status.
return 'P:{p} R: {r} D:{d}'.format(
p=colored(_status_color('pending'), self.children['Pending']),
r=colored(_status_color('running'), self.children['Running']),
d=colored(_status_color('dead'), self.children['Dead'])
)
def info(self, indent=''):
msg = (
indent
+ colored('bold', self.token)
+ ' '
+ _timetoiso(self.allocation_time())
+ ' : '
)
if self.pending():
msg += colored('blue', 'pending')
elif self.running():
msg += (
colored('yellow', 'running')
+ ' since {t}'.format(t=_timetoiso(self.start_time()))
)
elif self.success():
msg += (
colored('green', 'success')
+ ' (ran from {t1} to {t2})'.format(
t1=_timetoiso(self.start_time()),
c=colored('green', self.summary['Complete']),
s=colored('blue', self.summary['Starting']),
f=colored('red', self.summary['Failed']),
l=colored('orange', self.summary['Lost']), # noqa: E741
q=colored('pink', self.summary['Queued']),
r=colored('yellow', self.summary['Running'])
)
else:
# For 'periodic', 'trigger', the summary is empty;
# it would be interesting to show the summary of the
# dispatch children jobs as part of the corresponding
# execution status.
return 'P:{p} R: {r} D:{d}'.format(
p=colored(_status_color('pending'), self.children['Pending']),
r=colored(_status_color('running'), self.children['Running']),
d=colored(_status_color('dead'), self.children['Dead'])
)