Source code for galaxy.webapps.galaxy.api.tool_data

import os

import galaxy.queue_worker
from galaxy import (
    exceptions,
    web
)
from galaxy.web import (
    expose_api,
    expose_api_raw
)
from galaxy.web.base.controller import BaseAPIController


[docs]class ToolData(BaseAPIController): """ RESTful controller for interactions with tool data """
[docs] @web.require_admin @expose_api def index(self, trans, **kwds): """ GET /api/tool_data: returns a list tool_data tables:: """ return list(a.to_dict() for a in self._data_tables.values())
[docs] @web.require_admin @expose_api def show(self, trans, id, **kwds): return self._data_table(id).to_dict(view='element')
[docs] @web.require_admin @expose_api def reload(self, trans, id, **kwd): """ GET /api/tool_data/{id}/reload Reloads a tool_data table. """ decoded_tool_data_id = id data_table = trans.app.tool_data_tables.data_tables.get(decoded_tool_data_id) data_table.reload_from_files() galaxy.queue_worker.send_control_task(trans.app, 'reload_tool_data_tables', noop_self=True, kwargs={'table_name': decoded_tool_data_id}) return self._data_table(decoded_tool_data_id).to_dict(view='element')
[docs] @web.require_admin @expose_api def delete(self, trans, id, **kwd): """ DELETE /api/tool_data/{id} Removes an item from a data table :type id: str :param id: the id of the data table containing the item to delete :type kwd: dict :param kwd: (required) dictionary structure containing: * payload: a dictionary itself containing: * values: <TAB> separated list of column contents, there must be a value for all the columns of the data table """ decoded_tool_data_id = id try: data_table = trans.app.tool_data_tables.data_tables.get(decoded_tool_data_id) except Exception: data_table = None if not data_table: trans.response.status = 400 return "Invalid data table id ( %s ) specified." % str(decoded_tool_data_id) values = None if kwd.get('payload', None): values = kwd['payload'].get('values', '') if not values: trans.response.status = 400 return "Invalid data table item ( %s ) specified." % str(values) split_values = values.split("\t") if len(split_values) != len(data_table.get_column_name_list()): trans.response.status = 400 return "Invalid data table item ( %s ) specified. Wrong number of columns (%s given, %s required)." % (str(values), str(len(split_values)), str(len(data_table.get_column_name_list()))) data_table.remove_entry(split_values) galaxy.queue_worker.send_control_task(trans.app, 'reload_tool_data_tables', noop_self=True, kwargs={'table_name': decoded_tool_data_id}) return self._data_table(decoded_tool_data_id).to_dict(view='element')
[docs] @web.require_admin @expose_api def show_field(self, trans, id, value, **kwds): """ GET /api/tool_data/<id>/fields/<value> Get information about a partiular field in a tool_data table """ return self._data_table_field(id, value).to_dict()
[docs] @web.require_admin @expose_api_raw def download_field_file(self, trans, id, value, path, **kwds): field_value = self._data_table_field(id, value) base_dir = field_value.get_base_dir() full_path = os.path.join(base_dir, path) if full_path not in field_value.get_files(): raise exceptions.ObjectNotFound("No such path in data table field.") return open(full_path, "rb")
def _data_table_field(self, id, value): out = self._data_table(id).get_field(value) if out is None: raise exceptions.ObjectNotFound("No such field %s in data table %s." % (value, id)) return out def _data_table(self, id): try: return self._data_tables[id] except IndexError: raise exceptions.ObjectNotFound("No such data table %s" % id) @property def _data_tables(self): return self.app.tool_data_tables.data_tables