diff --git a/lib/galaxy/managers/display_applications.py b/lib/galaxy/managers/display_applications.py new file mode 100644 index 00000000000..2da27193768 --- /dev/null +++ b/lib/galaxy/managers/display_applications.py @@ -0,0 +1,61 @@ +import logging +from typing import ( + Any, Dict, + List, +) + +from galaxy.datatypes.registry import Registry + +log = logging.getLogger(__name__) + + +class DisplayApplicationsManager: + """Interface/service object for sharing logic between controllers.""" + + def __init__(self, app): + self._app = app + + @property + def datatypes_registry(self) -> Registry: + return self._app.datatypes_registry + + def index(self) -> List[Any]: + """ + Returns the list of display applications. + + :returns: list of available display applications + :rtype: list + """ + rval = [] + for display_app in self.datatypes_registry.display_applications.values(): + rval.append({ + 'id': display_app.id, + 'name': display_app.name, + 'version': display_app.version, + 'filename_': display_app._filename, + 'links': [{'name': link.name} for link in display_app.links.values()] + }) + return rval + + def reload(self, ids: List[str]) -> Dict[str, Any]: + """ + Reloads the list of display applications. + + :param ids: list containing ids of display to be reloaded + :type ids: list + """ + self._app.queue_worker.send_control_task( + 'reload_display_application', + noop_self=True, + kwargs={'display_application_ids': ids} + ) + reloaded, failed = self.datatypes_registry.reload_display_applications(ids) + if not reloaded and failed: + message = 'Unable to reload any of the %i requested display applications ("%s").' % (len(failed), '", "'.join(failed)) + elif failed: + message = 'Reloaded %i display applications ("%s"), but failed to reload %i display applications ("%s").' % (len(reloaded), '", "'.join(reloaded), len(failed), '", "'.join(failed)) + elif not reloaded: + message = 'You need to request at least one display application to reload.' + else: + message = 'Reloaded %i requested display applications ("%s").' % (len(reloaded), '", "'.join(reloaded)) + return {'message': message, 'reloaded': reloaded, 'failed': failed} diff --git a/lib/galaxy/webapps/galaxy/api/display_applications.py b/lib/galaxy/webapps/galaxy/api/display_applications.py index cdba5b8047d..56adeaf3ae3 100644 --- a/lib/galaxy/webapps/galaxy/api/display_applications.py +++ b/lib/galaxy/webapps/galaxy/api/display_applications.py @@ -3,6 +3,8 @@ API operations on annotations. """ import logging +from galaxy.app import StructuredApp +from galaxy.managers.display_applications import DisplayApplicationsManager from galaxy.web import ( expose_api, require_admin, @@ -14,6 +16,10 @@ log = logging.getLogger(__name__) class DisplayApplicationsController(BaseAPIController): + def __init__(self, app: StructuredApp): + super().__init__(app) + self.manager = DisplayApplicationsManager(app) + @expose_api def index(self, trans, **kwd): """ @@ -24,16 +30,7 @@ class DisplayApplicationsController(BaseAPIController): :returns: list of available display applications :rtype: list """ - response = [] - for display_app in trans.app.datatypes_registry.display_applications.values(): - response.append({ - 'id': display_app.id, - 'name': display_app.name, - 'version': display_app.version, - 'filename_': display_app._filename, - 'links': [{'name': link.name} for link in display_app.links.values()] - }) - return response + return self.manager.index() @expose_api @require_admin @@ -47,19 +44,5 @@ class DisplayApplicationsController(BaseAPIController): :type ids: list """ payload = payload or {} - ids = payload.get('ids') - trans.app.queue_worker.send_control_task( - 'reload_display_application', - noop_self=True, - kwargs={'display_application_ids': ids} - ) - reloaded, failed = trans.app.datatypes_registry.reload_display_applications(ids) - if not reloaded and failed: - message = 'Unable to reload any of the %i requested display applications ("%s").' % (len(failed), '", "'.join(failed)) - elif failed: - message = 'Reloaded %i display applications ("%s"), but failed to reload %i display applications ("%s").' % (len(reloaded), '", "'.join(reloaded), len(failed), '", "'.join(failed)) - elif not reloaded: - message = 'You need to request at least one display application to reload.' - else: - message = 'Reloaded %i requested display applications ("%s").' % (len(reloaded), '", "'.join(reloaded)) - return {'message': message, 'reloaded': reloaded, 'failed': failed} + ids = payload.get('ids', []) + return self.manager.reload(ids)