Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@app.callback(Output('page_content', 'children'),
[Input('url', 'pathname')])
def display_page(pathname):
if pathname == '/': # TODO Redirect to workflows or show special page
pass
elif pathname == '/workflows':
return workflows.layout
elif '/workflows' in str(pathname):
pathname_split = pathname.split('/')
run_id = pathname_split[pathname_split.index('workflows') + 1]
if run_id and re.fullmatch(RUN_ID_REGEX, run_id):
return nav_bar.display_workflow(run_id=run_id)
else:
return'Invalid run id'
else:
return '404'
@app.callback(Output('tasks_details', 'children'),
[Input('run_id', 'children')],
[State('url', 'pathname')])
def tasks_details(run_id, pathname):
sql_conn = get_db()
df_task = pd.read_sql_query('SELECT task_id, task_func_name FROM task WHERE run_id=(?)',
sql_conn, params=(run_id,))
close_db()
tasks = []
for task_id in df_task['task_id']:
tasks.append({'label': task_id, 'value': task_id})
return [html.P(className='view_title', children='Tasks view'),
html.Div(id='select_task_id',
children=[
html.H4('Select Task ID'),
@app.callback(Output('workflow_details', 'children'),
[Input('run_number_dropdown', 'value')])
def workflow_details(run_id):
return [html.A(id='run_id', children=run_id, hidden=True)] + [plot.html for plot in plots]
@app.callback(Output('tabs-content', 'children'),
[Input('tabs', 'value')])
def render_content(tab):
if tab == 'workflow':
return workflow_details.layout
elif tab == 'tasks':
return tasks_details.layout
@app.callback(Output('tasks_per_app_plot_tasks', 'figure'),
[Input('apps_dropdown', 'value')],
[State('run_number_dropdown', 'value')])
def tasks_per_app_plot_callback(apps, run_id):
if type(apps) is dict:
apps = ['', apps['label']]
elif len(apps) == 1:
apps.append('')
sql_conn = get_db()
df_status = pd.read_sql_query('SELECT run_id, task_id, task_status_name, timestamp FROM task_status WHERE run_id=(?)',
sql_conn, params=(run_id, ))
df_task = pd.read_sql_query('SELECT task_id, task_func_name FROM task WHERE run_id=(?) AND task_fn_hash IN {apps}'.format(apps=tuple(apps)),
sql_conn, params=(run_id, ))
close_db()
def y_axis_setup(array):
@app.callback(Output('apps_details', 'children'),
[Input('run_id', 'children')])
def tasks_details(run_id):
sql_conn = get_db()
df_task = pd.read_sql_query('SELECT task_id, task_func_name FROM task WHERE run_id=(?)',
sql_conn, params=(run_id,))
close_db()
apps = []
for _app in df_task['task_func_name'].unique():
apps.append(dict(label=_app, value=_app))
return [html.P(className='view_title', children='Apps view'),
html.Div(id='select_apps',
children=[
html.H4('Select apps'),
dcc.Dropdown(
@app.callback(Output('workflow_details', 'children'),
[Input('run_id', 'children')])
def workflow_details(run_id):
return [html.P(className='view_title', children='Workflow view')] + [plot.html(run_id) for plot in plots]
@app.callback(Output('nav_content', 'children'),
[Input('url', 'pathname')])
def render_content(pathname):
if '/workflow-view' in str(pathname):
return workflow_details.layout
elif '/apps-view' in str(pathname):
return apps_details.layout
elif '/tasks-view' in str(pathname):
return tasks_details.layout
else:
return html.H2('Select a view from the menu above', style=dict(textAlign='center'))
@app.callback(Output(plot_id, 'figure'), plot_input_args, plot_state_args)
def plot_callback(*args):
return plot(*args)