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 dm: dm[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": f"Invalid Data Manager ({data_manager_id}) was requested", "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(f"Bad job id ({job_id}) passed to job_info: {e}") if not job: return {"message": f"Invalid job ({job_id}) was requested", "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(f"Unable to obtain data_table info for hda ({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": f"Invalid data table '{data_table_name}' was requested.", "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": f"Invalid Data table ({table_name}) was requested", "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=f'The data table "{table_name}" has been reloaded.', status="done", ) return trans.response.send_redirect(redirect_url)