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.managers.display_applications
import logging
from typing import Optional
from urllib.parse import unquote_plus
from pydantic import BaseModel
from galaxy.datatypes.display_applications.util import (
decode_dataset_user,
encode_dataset_user,
)
from galaxy.datatypes.registry import Registry
from galaxy.exceptions import MessageException
from galaxy.managers.context import ProvidesUserContext
from galaxy.structured_app import StructuredApp
log = logging.getLogger(__name__)
[docs]
class CreateLinkStep(BaseModel):
name: str
state: Optional[str] = None
ready: Optional[bool] = False
[docs]
class CreateLinkFeedback(BaseModel):
messages: Optional[list[tuple[str, str]]] = None
refresh: Optional[bool] = False
resource: Optional[str] = None
preparable_steps: Optional[list[CreateLinkStep]] = None
[docs]
class CreateLinkIncoming(BaseModel):
app_name: str
dataset_id: str
link_name: str
kwd: Optional[dict[str, str]] = None
[docs]
class DisplayApplication(BaseModel):
id: str
name: str
version: str
filename_: str
links: list[Link]
[docs]
class ReloadFeedback(BaseModel):
message: str
reloaded: list[Optional[str]]
failed: list[Optional[str]]
[docs]
class DisplayApplicationsManager:
"""Interface/service object for sharing logic between controllers."""
@property
def datatypes_registry(self) -> Registry:
return self._app.datatypes_registry
[docs]
def index(self) -> list[DisplayApplication]:
"""
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(
DisplayApplication(
id=display_app.id,
name=display_app.name,
version=display_app.version,
filename_=display_app._filename,
links=[Link(name=link.name) for link in display_app.links.values()],
)
)
return rval
[docs]
def reload(self, ids: list[str]) -> ReloadFeedback:
"""
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 {} requested display applications ("{}").'.format(
len(failed),
'", "'.join(failed),
)
elif failed:
message = (
'Reloaded {} display applications ("{}"), but failed to reload {} display applications ("{}").'.format(
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 {} requested display applications ("{}").'.format(len(reloaded), '", "'.join(reloaded))
return ReloadFeedback(message=message, reloaded=reloaded, failed=failed)
[docs]
def create_link(
self,
trans: ProvidesUserContext,
app_name: str,
dataset_id: str,
link_name: str,
user_id: Optional[str] = None,
**kwds,
) -> CreateLinkFeedback:
"""Access to external display applications"""
if None in [app_name, link_name]:
raise MessageException("A display application name and link name must be provided.")
app_name = unquote_plus(app_name)
link_name = unquote_plus(link_name)
# Build list of parameters to pass in to display application logic (app_kwds)
app_kwds = {}
for name, value in dict(kwds).items(): # clone kwds because we remove stuff as we go.
if name.startswith("app_"):
app_kwds[name[len("app_") :]] = value
del kwds[name]
if kwds:
log.debug(f"Unexpected Keywords passed to display_application: {kwds}") # route memory?
# decode ids
data, user = decode_dataset_user(trans, dataset_id, user_id)
if not data:
raise MessageException(f"Invalid reference dataset id: {str(dataset_id)}.")
if user is None:
user = trans.user
if user:
user_roles = user.all_roles()
else:
user_roles = []
# Decode application name and link name
if self._can_access_dataset(trans, data, additional_roles=user_roles):
messages = []
preparable_steps = []
refresh = False
display_app = self.datatypes_registry.display_applications.get(app_name)
if not display_app:
log.debug("Unknown display application has been requested: %s", app_name)
raise MessageException(f"The requested display application ({app_name}) is not available.")
dataset_hash, user_hash = encode_dataset_user(trans, data, user)
try:
display_link = display_app.get_link(link_name, data, dataset_hash, user_hash, trans, app_kwds)
except Exception as e:
log.debug("Error generating display_link: %s", e)
# User can sometimes recover from, e.g. conversion errors by fixing input metadata, so use conflict
raise MessageException(f"Error generating display_link: {e}")
if not display_link:
log.debug("Unknown display link has been requested: %s", link_name)
raise MessageException(f"Unknown display link has been requested: {link_name}")
if data.state == data.states.ERROR:
messages.append(
(
"This dataset is in an error state, you cannot view it at an external display application.",
"error",
)
)
elif data.deleted:
messages.append(
("This dataset has been deleted, you cannot view it at an external display application.", "error")
)
elif data.state != data.states.OK:
messages.append(
(
"You must wait for this dataset to be created before you can view it at an external display application.",
"info",
)
)
refresh = True
else:
# We have permissions, dataset is not deleted and is in OK state, allow access
if display_link.display_ready():
# redirect user to url generated by display link
return CreateLinkFeedback(resource=display_link.display_url())
else:
refresh = True
messages.append(
(
"Launching this display application required additional datasets to be generated, you can view the status of these jobs below.",
"info",
)
)
if not display_link.preparing_display():
display_link.prepare_display()
preparable_steps = display_link.get_prepare_steps()
return CreateLinkFeedback(
messages=messages,
refresh=refresh,
preparable_steps=preparable_steps,
)
raise MessageException("You do not have permission to view this dataset at an external display application.")
def _can_access_dataset(self, trans, dataset_association, allow_admin=True, additional_roles=None):
roles = trans.get_current_user_roles()
if additional_roles:
roles = roles + additional_roles
return (allow_admin and trans.user_is_admin) or trans.app.security_agent.can_access_dataset(
roles, dataset_association.dataset
)