rattail/tailbone
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Warn user when luigi is not installed, for relevant view

Browse source 
better than getting a server error
This commit is contained in:
Lance Edgar 2022-12-23 18:55:53 -06:00
parent 6fbc79fe5e
commit 8a6fdb5ea5
1 changed files with 22 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
tailbone/views/luigi.py
View file

@ -62,9 +62,22 @@ class LuigiTaskView(MasterView):
def __init__(self, request, context=None): def __init__(self, request, context=None):
super(LuigiTaskView, self).__init__(request, context=context) super(LuigiTaskView, self).__init__(request, context=context)
app = self.get_rattail_app() app = self.get_rattail_app()
# nb. luigi may not be installed, which (for now) may prevent
# us from getting our handler; in which case warn user
try:
self.luigi_handler = app.get_luigi_handler() self.luigi_handler = app.get_luigi_handler()
except Exception as error:
self.luigi_handler = None
self.luigi_handler_error = error
log.warning("could not get luigi handler", exc_info=True)
def index(self): def index(self):
if not self.luigi_handler:
self.request.session.flash("Could not create handler: {}".format(
simple_error(self.luigi_handler_error)), 'error')
luigi_url = self.rattail_config.get('rattail.luigi', 'url') luigi_url = self.rattail_config.get('rattail.luigi', 'url')
history_url = '{}/history'.format(luigi_url.rstrip('/')) if luigi_url else None history_url = '{}/history'.format(luigi_url.rstrip('/')) if luigi_url else None
return self.render_to_response('index', { return self.render_to_response('index', {
@ -147,14 +160,20 @@ class LuigiTaskView(MasterView):
return context return context
def get_overnight_tasks(self): def get_overnight_tasks(self):
if self.luigi_handler:
tasks = self.luigi_handler.get_all_overnight_tasks() tasks = self.luigi_handler.get_all_overnight_tasks()
else:
tasks = []
for task in tasks: for task in tasks:
if task['last_date']: if task['last_date']:
task['last_date'] = six.text_type(task['last_date']) task['last_date'] = six.text_type(task['last_date'])
return tasks return tasks
def get_backfill_tasks(self): def get_backfill_tasks(self):
if self.luigi_handler:
tasks = self.luigi_handler.get_all_backfill_tasks() tasks = self.luigi_handler.get_all_backfill_tasks()
else:
tasks = []
for task in tasks: for task in tasks:
if task['last_date']: if task['last_date']:
task['last_date'] = six.text_type(task['last_date']) task['last_date'] = six.text_type(task['last_date'])