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.api.visualizations
"""
Visualizations resource control over the API.
NOTE!: this is a work in progress and functionality and data structures
may change often.
"""
import json
import logging
from typing import Optional
from fastapi import (
Body,
Path,
Query,
Response,
status,
)
from typing_extensions import Annotated
from galaxy import (
exceptions,
util,
web,
)
from galaxy.managers.context import ProvidesUserContext
from galaxy.model.base import transaction
from galaxy.model.item_attrs import UsesAnnotations
from galaxy.schema.fields import DecodedDatabaseIdField
from galaxy.schema.schema import (
SetSlugPayload,
ShareWithPayload,
ShareWithStatus,
SharingStatus,
)
from galaxy.schema.visualization import (
VisualizationIndexQueryPayload,
VisualizationSortByEnum,
VisualizationSummaryList,
)
from galaxy.util.hash_util import md5_hash_str
from galaxy.web import expose_api
from galaxy.webapps.base.controller import UsesVisualizationMixin
from galaxy.webapps.base.webapp import GalaxyWebTransaction
from galaxy.webapps.galaxy.api import (
BaseGalaxyAPIController,
depends,
DependsOnTrans,
IndexQueryTag,
Router,
search_query_param,
)
from galaxy.webapps.galaxy.api.common import (
LimitQueryParam,
OffsetQueryParam,
)
from galaxy.webapps.galaxy.services.visualizations import VisualizationsService
log = logging.getLogger(__name__)
router = Router(tags=["visualizations"])
DeletedQueryParam: bool = Query(
default=False, title="Display deleted", description="Whether to include deleted visualizations in the result."
)
UserIdQueryParam: Optional[DecodedDatabaseIdField] = Query(
default=None,
title="Encoded user ID to restrict query to, must be own id if not an admin user",
)
query_tags = [
IndexQueryTag("title", "The visualization's title."),
IndexQueryTag("slug", "The visualization's slug.", "s"),
IndexQueryTag("tag", "The visualization's tags.", "t"),
IndexQueryTag("user", "The visualization's owner's username.", "u"),
]
SearchQueryParam: Optional[str] = search_query_param(
model_name="Visualization",
tags=query_tags,
free_text_fields=["title", "slug", "tag", "type"],
)
SharingQueryParam: bool = Query(
default=False, title="Provide sharing status", description="Whether to provide sharing details in the result."
)
ShowOwnQueryParam: bool = Query(default=True, title="Show visualizations owned by user.", description="")
ShowPublishedQueryParam: bool = Query(default=True, title="Include published visualizations.", description="")
ShowSharedQueryParam: bool = Query(
default=False, title="Include visualizations shared with authenticated user.", description=""
)
SortByQueryParam: VisualizationSortByEnum = Query(
default="update_time",
title="Sort attribute",
description="Sort visualization index by this specified attribute on the visualization model",
)
SortDescQueryParam: bool = Query(
default=True,
title="Sort Descending",
description="Sort in descending order?",
)
VisualizationIdPathParam = Annotated[
DecodedDatabaseIdField,
Path(..., title="Visualization ID", description="The encoded database identifier of the Visualization."),
]
[docs]@router.cbv
class FastAPIVisualizations:
service: VisualizationsService = depends(VisualizationsService)
[docs] @router.get(
"/api/visualizations",
summary="Returns visualizations for the current user.",
)
async def index(
self,
response: Response,
trans: ProvidesUserContext = DependsOnTrans,
deleted: bool = DeletedQueryParam,
limit: Optional[int] = LimitQueryParam,
offset: Optional[int] = OffsetQueryParam,
user_id: Optional[DecodedDatabaseIdField] = UserIdQueryParam,
show_own: bool = ShowOwnQueryParam,
show_published: bool = ShowPublishedQueryParam,
show_shared: bool = ShowSharedQueryParam,
sort_by: VisualizationSortByEnum = SortByQueryParam,
sort_desc: bool = SortDescQueryParam,
search: Optional[str] = SearchQueryParam,
) -> VisualizationSummaryList:
payload = VisualizationIndexQueryPayload.model_construct(
deleted=deleted,
user_id=user_id,
show_published=show_published,
show_own=show_own,
show_shared=show_shared,
sort_by=sort_by,
sort_desc=sort_desc,
limit=limit,
offset=offset,
search=search,
)
entries, total_matches = self.service.index(trans, payload, include_total_count=True)
response.headers["total_matches"] = str(total_matches)
return entries
[docs] @router.get(
"/api/visualizations/{id}/sharing",
summary="Get the current sharing status of the given Visualization.",
)
def sharing(
self,
id: VisualizationIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
) -> SharingStatus:
"""Return the sharing status of the item."""
return self.service.shareable_service.sharing(trans, id)
[docs] @router.put(
"/api/visualizations/{id}/enable_link_access",
summary="Makes this item accessible by a URL link.",
)
def enable_link_access(
self,
id: VisualizationIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
) -> SharingStatus:
"""Makes this item accessible by a URL link and return the current sharing status."""
return self.service.shareable_service.enable_link_access(trans, id)
[docs] @router.put(
"/api/visualizations/{id}/disable_link_access",
summary="Makes this item inaccessible by a URL link.",
)
def disable_link_access(
self,
id: VisualizationIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
) -> SharingStatus:
"""Makes this item inaccessible by a URL link and return the current sharing status."""
return self.service.shareable_service.disable_link_access(trans, id)
[docs] @router.put(
"/api/visualizations/{id}/publish",
summary="Makes this item public and accessible by a URL link.",
)
def publish(
self,
id: VisualizationIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
) -> SharingStatus:
"""Makes this item publicly available by a URL link and return the current sharing status."""
return self.service.shareable_service.publish(trans, id)
[docs] @router.put(
"/api/visualizations/{id}/unpublish",
summary="Removes this item from the published list.",
)
def unpublish(
self,
id: VisualizationIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
) -> SharingStatus:
"""Removes this item from the published list and return the current sharing status."""
return self.service.shareable_service.unpublish(trans, id)
[docs] @router.put(
"/api/visualizations/{id}/slug",
summary="Set a new slug for this shared item.",
status_code=status.HTTP_204_NO_CONTENT,
)
def set_slug(
self,
id: VisualizationIdPathParam,
trans: ProvidesUserContext = DependsOnTrans,
payload: SetSlugPayload = Body(...),
):
"""Sets a new slug to access this item by URL. The new slug must be unique."""
self.service.shareable_service.set_slug(trans, id, payload)
return Response(status_code=status.HTTP_204_NO_CONTENT)
[docs]class VisualizationsController(BaseGalaxyAPIController, UsesVisualizationMixin, UsesAnnotations):
"""
RESTful controller for interactions with visualizations.
"""
service: VisualizationsService = depends(VisualizationsService)
[docs] @expose_api
def show(self, trans: GalaxyWebTransaction, id: str, **kwargs):
"""
GET /api/visualizations/{viz_id}
"""
# TODO: revisions should be a contents/nested controller like viz/xxx/r/xxx)?
# the important thing is the config
# TODO:?? /api/visualizations/registry -> json of registry.listings?
visualization = self.get_visualization(trans, id, check_ownership=False, check_accessible=True)
dictionary = trans.security.encode_dict_ids(self.get_visualization_dict(visualization))
dictionary["url"] = web.url_for(
controller="visualization",
action="display_by_username_and_slug",
username=visualization.user.username,
slug=visualization.slug,
)
dictionary["username"] = visualization.user.username
dictionary["email_hash"] = md5_hash_str(visualization.user.email)
dictionary["tags"] = visualization.make_tag_string_list()
dictionary["annotation"] = self.get_item_annotation_str(trans.sa_session, trans.user, visualization)
# need to encode ids in revisions as well
encoded_revisions = []
for revision in dictionary["revisions"]:
# NOTE: does not encode ids inside the configs
encoded_revisions.append(trans.security.encode_id(revision))
dictionary["revisions"] = encoded_revisions
dictionary["latest_revision"] = trans.security.encode_dict_ids(dictionary["latest_revision"])
if trans.app.visualizations_registry:
visualization = trans.app.visualizations_registry.get_plugin(dictionary["type"])
dictionary["plugin"] = visualization.to_dict()
return dictionary
[docs] @expose_api
def create(self, trans: GalaxyWebTransaction, payload: dict, **kwargs):
"""
POST /api/visualizations
creates a new visualization using the given payload
POST /api/visualizations?import_id={encoded_visualization_id}
imports a copy of an existing visualization into the user's workspace
"""
rval = None
if "import_id" in payload:
import_id = payload["import_id"]
visualization = self.import_visualization(trans, import_id, user=trans.user)
else:
payload = self._validate_and_parse_payload(payload)
# must have a type (I've taken this to be the visualization name)
if "type" not in payload:
raise exceptions.RequestParameterMissingException("key/value 'type' is required")
vis_type = payload.pop("type", False)
payload["save"] = True
try:
# generate defaults - this will err if given a weird key?
visualization = self.create_visualization(trans, vis_type, **payload)
except ValueError as val_err:
raise exceptions.RequestParameterMissingException(str(val_err))
rval = {"id": trans.security.encode_id(visualization.id)}
return rval
[docs] @expose_api
def update(self, trans: GalaxyWebTransaction, id: str, payload: dict, **kwargs):
"""
PUT /api/visualizations/{encoded_visualization_id}
"""
rval = None
payload = self._validate_and_parse_payload(payload)
# there's a differentiation here between updating the visualization and creating a new revision
# that needs to be handled clearly here or alternately, using a different controller
# like e.g. PUT /api/visualizations/{id}/r/{id}
# TODO: consider allowing direct alteration of revisions title (without a new revision)
# only create a new revsion on a different config
# only update owned visualizations
visualization = self.get_visualization(trans, id, check_ownership=True)
title = payload.get("title", visualization.latest_revision.title)
dbkey = payload.get("dbkey", visualization.latest_revision.dbkey)
deleted = payload.get("deleted", visualization.deleted)
config = payload.get("config", visualization.latest_revision.config)
latest_config = visualization.latest_revision.config
if (
(title != visualization.latest_revision.title)
or (dbkey != visualization.latest_revision.dbkey)
or (json.dumps(config) != json.dumps(latest_config))
):
revision = self.add_visualization_revision(trans, visualization, config, title, dbkey)
rval = {"id": id, "revision": revision.id}
# allow updating vis title
visualization.title = title
visualization.deleted = deleted
with transaction(trans.sa_session):
trans.sa_session.commit()
return rval
def _validate_and_parse_payload(self, payload):
"""
Validate and parse incomming data payload for a visualization.
"""
# This layer handles (most of the stricter idiot proofing):
# - unknown/unallowed keys
# - changing data keys from api key to attribute name
# - protection against bad data form/type
# - protection against malicious data content
# all other conversions and processing (such as permissions, etc.) should happen down the line
# keys listed here don't error when attempting to set, but fail silently
# this allows PUT'ing an entire model back to the server without attribute errors on uneditable attrs
valid_but_uneditable_keys = (
"id",
"model_class",
# TODO: fill out when we create to_dict, get_dict, whatevs
)
# TODO: importable
ValidationError = exceptions.RequestParameterInvalidException
validated_payload = {}
for key, val in payload.items():
# TODO: validate types in VALID_TYPES/registry names at the mixin/model level?
if key == "type":
if not isinstance(val, str):
raise ValidationError(f"{key} must be a string or unicode: {str(type(val))}")
val = util.sanitize_html.sanitize_html(val)
elif key == "config":
if not isinstance(val, dict):
raise ValidationError(f"{key} must be a dictionary: {str(type(val))}")
elif key == "annotation":
if not isinstance(val, str):
raise ValidationError(f"{key} must be a string or unicode: {str(type(val))}")
val = util.sanitize_html.sanitize_html(val)
elif key == "deleted":
if not isinstance(val, bool):
raise ValidationError(f"{key} must be a bool: {str(type(val))}")
# these are keys that actually only be *updated* at the revision level and not here
# (they are still valid for create, tho)
elif key == "title":
if not isinstance(val, str):
raise ValidationError(f"{key} must be a string or unicode: {str(type(val))}")
val = util.sanitize_html.sanitize_html(val)
elif key == "slug":
if not isinstance(val, str):
raise ValidationError(f"{key} must be a string: {str(type(val))}")
val = util.sanitize_html.sanitize_html(val)
elif key == "dbkey":
if not isinstance(val, str):
raise ValidationError(f"{key} must be a string or unicode: {str(type(val))}")
val = util.sanitize_html.sanitize_html(val)
elif key not in valid_but_uneditable_keys:
continue
# raise AttributeError( 'unknown key: %s' %( str( key ) ) )
validated_payload[key] = val
return validated_payload