Warning

This document is for an in-development version of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.webapps.galaxy.controllers.data_manager

import logging
from json import loads

import paste.httpexceptions

from galaxy import web
from galaxy.util import nice_size, unicodify
from galaxy.webapps.base.controller import BaseUIController

log = logging.getLogger(__name__)


[docs]class DataManager(BaseUIController):
[docs] @web.expose @web.json def data_managers_list(self, trans, **kwd): not_is_admin = not trans.user_is_admin if not_is_admin and not trans.app.config.enable_data_manager_user_view: raise paste.httpexceptions.HTTPUnauthorized("This Galaxy instance is not configured to allow non-admins to view the data manager.") message = kwd.get('message', '') status = kwd.get('status', 'info') data_managers = [] for data_manager_id, data_manager in sorted(trans.app.data_managers.data_managers.items(), key=lambda data_manager: data_manager[1].name): data_managers.append({'toolUrl': web.url_for(controller='root', tool_id=data_manager.tool.id), 'id': data_manager_id, 'name': data_manager.name, 'description': data_manager.description.lower()}) data_tables = [] managed_table_names = trans.app.data_managers.managed_data_tables.keys() for table_name in sorted(trans.app.tool_data_tables.get_tables().keys()): data_tables.append({'name': table_name, 'managed': True if table_name in managed_table_names else False}) return {'dataManagers': data_managers, 'dataTables': data_tables, 'viewOnly': not_is_admin, 'message': message, 'status': status}
[docs] @web.expose @web.json def jobs_list(self, trans, **kwd): not_is_admin = not trans.user_is_admin if not_is_admin and not trans.app.config.enable_data_manager_user_view: raise paste.httpexceptions.HTTPUnauthorized("This Galaxy instance is not configured to allow non-admins to view the data manager.") message = kwd.get('message', '') status = kwd.get('status', 'info') data_manager_id = kwd.get('id', None) data_manager = trans.app.data_managers.get_manager(data_manager_id) if data_manager is None: return {'message': "Invalid Data Manager (%s) was requested" % data_manager_id, 'status': "error"} jobs = [] for assoc in trans.sa_session.query(trans.app.model.DataManagerJobAssociation).filter_by(data_manager_id=data_manager_id): j = assoc.job jobs.append({ 'id': j.id, 'encId': trans.security.encode_id(j.id), 'runUrl': web.url_for(controller="tool_runner", action="rerun", job_id=trans.security.encode_id(j.id)), 'user': j.history.user.email if j.history and j.history.user else "anonymous", 'updateTime': j.update_time.isoformat(), 'state': j.state, 'commandLine': j.command_line, 'jobRunnerName': j.job_runner_name, 'jobRunnerExternalId': j.job_runner_external_id }) jobs.reverse() return {'dataManager': {'name': data_manager.name, 'description': data_manager.description.lower(), 'toolUrl': web.url_for(controller='root', tool_id=data_manager.tool.id)}, 'jobs': jobs, 'viewOnly': not_is_admin, 'message': message, 'status': status}
[docs] @web.expose @web.json def job_info(self, trans, **kwd): not_is_admin = not trans.user_is_admin if not_is_admin and not trans.app.config.enable_data_manager_user_view: raise paste.httpexceptions.HTTPUnauthorized("This Galaxy instance is not configured to allow non-admins to view the data manager.") message = kwd.get('message', '') status = kwd.get('status', 'info') job_id = kwd.get('id', None) try: job_id = trans.security.decode_id(job_id) job = trans.sa_session.query(trans.app.model.Job).get(job_id) except Exception as e: job = None log.error("Bad job id ({}) passed to job_info: {}".format(job_id, e)) if not job: return {'message': "Invalid job (%s) was requested" % job_id, 'status': "error"} data_manager_id = job.data_manager_association.data_manager_id data_manager = trans.app.data_managers.get_manager(data_manager_id) hdas = [assoc.dataset for assoc in job.get_output_datasets()] hda_info = [] data_manager_output = [] error_messages = [] for hda in hdas: hda_info.append({'id': hda.id, 'encId': trans.security.encode_id(hda.id), 'name': hda.name, 'created': unicodify(hda.create_time.strftime(trans.app.config.pretty_datetime_format)), 'fileSize': nice_size(hda.dataset.file_size), 'fileName': hda.file_name, 'infoUrl': web.url_for(controller='dataset', action='show_params', dataset_id=trans.security.encode_id(hda.id))}) try: data_manager_json = loads(open(hda.get_file_name()).read()) except Exception as e: data_manager_json = {} error_messages.append("Unable to obtain data_table info for hda ({}): {}".format(hda.id, e)) values = [] for key, value in data_manager_json.get('data_tables', {}).items(): values.append((key, value)) data_manager_output.append(values) return {'jobId': job_id, 'exitCode': job.exit_code, 'runUrl': web.url_for(controller="tool_runner", action="rerun", job_id=trans.security.encode_id(job.id)), 'commandLine': job.command_line, 'dataManager': {'id': data_manager_id, 'name': data_manager.name, 'description': data_manager.description.lower(), 'toolUrl': web.url_for(controller='root', tool_id=data_manager.tool.id)}, 'hdaInfo': hda_info, 'dataManagerOutput': data_manager_output, 'errorMessages': error_messages, 'viewOnly': not_is_admin, 'message': message, 'status': status}
[docs] @web.expose @web.json def tool_data_table_info(self, trans, **kwd): return self.tool_data_table_info_1(trans, **kwd)
[docs] def tool_data_table_info_1(self, trans, **kwd): not_is_admin = not trans.user_is_admin if not_is_admin and not trans.app.config.enable_data_manager_user_view: raise paste.httpexceptions.HTTPUnauthorized("This Galaxy instance is not configured to allow non-admins to view the data manager.") message = kwd.get('message', '') status = kwd.get('status', 'info') data_table_name = kwd.get('table_name', None) if not data_table_name: return {'message': "No data table was requested.", 'status': "error"} data_table = trans.app.tool_data_tables.get(data_table_name, None) if data_table is None: return {'message': "Invalid data table '%s' was requested." % data_table_name, 'status': "error"} return {'dataTable': {'name': data_table.name, 'columns': data_table.get_column_name_list(), 'data': data_table.data}, 'viewOnly': not_is_admin, 'message': message, 'status': status}
[docs] @web.expose @web.json @web.require_admin def reload_tool_data_tables(self, trans, table_name=None, **kwd): if table_name and isinstance(table_name, str): table_name = table_name.split(",") # Reload the tool data tables table_names = self.app.tool_data_tables.reload_tables(table_names=table_name) trans.app.queue_worker.send_control_task( 'reload_tool_data_tables', noop_self=True, kwargs={'table_name': table_name} ) data = None if table_names: message = "Reloaded data table{} '{}'.".format('s'[len(table_names) == 1:], ', '.join(table_names)) data = self.tool_data_table_info_1(trans, table_name=table_names[0], message=message, status="done") else: data = {'message': "No data tables have been reloaded.", 'status': "error"} return data
[docs] @web.expose @web.json @web.require_admin def tool_data_table_items(self, trans, **kwd): data = {'columns': [], 'items': []} message = kwd.get('message', '') status = kwd.get('status', 'info') table_name = kwd.get('table_name', None) if not table_name: return { 'data': data, 'message': 'No Data table name provided.', 'status': 'warning', } data_table = trans.app.tool_data_tables.get(table_name, None) if data_table is None: return { 'data': data, 'message': 'Invalid Data table (%s) was requested' % table_name, 'status': 'error' } columns = data_table.get_column_name_list() rows = [dict(zip(columns, table_row)) for table_row in data_table.data] data['columns'] = columns data['items'] = rows return {'data': data, 'message': message, 'status': status}
[docs] @web.expose @web.json @web.require_admin def reload_tool_data_table(self, trans, **kwd): table_name = kwd.get('table_name', None) if not table_name: return { 'message': 'No data table has been reloaded.', 'status': 'error', } redirect_url = web.url_for( controller='data_manager', action='tool_data_table_items', table_name=table_name, message='The data table "%s" has been reloaded.' % table_name, status='done', ) return trans.response.send_redirect(redirect_url)