from galaxy import (
model,
web,
)
from galaxy.managers.hdas import HDAManager
from galaxy.managers.histories import (
HistoryManager,
HistorySerializer,
)
from galaxy.managers.pages import (
get_page as get_page_,
page_exists,
PageManager,
)
from galaxy.managers.sharable import SlugBuilder
from galaxy.managers.users import get_user_by_username
from galaxy.managers.workflows import WorkflowsManager
from galaxy.model.base import transaction
from galaxy.model.item_attrs import UsesItemRatings
from galaxy.schema.schema import CreatePagePayload
from galaxy.structured_app import StructuredApp
from galaxy.util.sanitize_html import sanitize_html
from galaxy.web import error
from galaxy.webapps.base.controller import (
BaseUIController,
SharableMixin,
UsesStoredWorkflowMixin,
UsesVisualizationMixin,
)
from galaxy.webapps.galaxy.api import depends
# Adapted from the _BaseHTMLProcessor class of https://github.com/kurtmckee/feedparser
[docs]class PageController(BaseUIController, SharableMixin, UsesStoredWorkflowMixin, UsesVisualizationMixin, UsesItemRatings):
page_manager: PageManager = depends(PageManager)
history_manager: HistoryManager = depends(HistoryManager)
history_serializer: HistorySerializer = depends(HistorySerializer)
hda_manager: HDAManager = depends(HDAManager)
workflow_manager: WorkflowsManager = depends(WorkflowsManager)
slug_builder: SlugBuilder = depends(SlugBuilder)
[docs] def __init__(self, app: StructuredApp):
super().__init__(app)
[docs] @web.expose_api
@web.require_login("create pages")
def create(self, trans, payload=None, **kwd):
"""
Create a new page.
"""
if trans.request.method == "GET":
form_title = "Create new Page"
title = ""
slug = ""
content = ""
content_hide = True
if "invocation_id" in kwd:
invocation_id = kwd.get("invocation_id")
form_title = f"{form_title} from Invocation Report"
slug = f"invocation-report-{invocation_id}"
invocation_report = self.workflow_manager.get_invocation_report(
trans, trans.security.decode_id(invocation_id)
)
title = invocation_report.get("title")
content = invocation_report.get("markdown")
content_hide = False
return {
"title": form_title,
"inputs": [
{
"name": "title",
"label": "Name",
"value": title,
},
{
"name": "slug",
"label": "Identifier",
"help": "A unique identifier that will be used for public links to this page. This field can only contain lowercase letters, numbers, and dashes (-).",
"value": slug,
},
{
"name": "annotation",
"label": "Annotation",
"help": "A description of the page. The annotation is shown alongside published pages.",
},
{
"name": "content_format",
"label": "Content Format",
"value": "markdown",
"hidden": True,
},
{
"name": "content",
"label": "Content",
"area": True,
"value": content,
"hidden": content_hide,
},
],
}
else:
page = self.page_manager.create_page(trans, CreatePagePayload(**payload))
return {"message": f"Page '{page.title}' successfully created.", "status": "success"}
[docs] @web.legacy_expose_api
@web.require_login("edit pages")
def edit(self, trans, payload=None, **kwd):
"""
Edit a page's attributes.
"""
id = kwd.get("id")
if not id:
return self.message_exception(trans, "No page id received for editing.")
decoded_id = self.decode_id(id)
user = trans.get_user()
p = trans.sa_session.get(model.Page, decoded_id)
p = self.security_check(trans, p, check_ownership=True)
if trans.request.method == "GET":
if p.slug is None:
self.slug_builder.create_item_slug(trans.sa_session, p)
return {
"title": "Edit page attributes",
"inputs": [
{"name": "title", "label": "Name", "value": p.title},
{
"name": "slug",
"label": "Identifier",
"value": p.slug,
"help": "A unique identifier that will be used for public links to this page. This field can only contain lowercase letters, numbers, and dashes (-).",
},
{
"name": "annotation",
"label": "Annotation",
"value": self.get_item_annotation_str(trans.sa_session, user, p),
"help": "A description of the page. The annotation is shown alongside published pages.",
},
],
}
else:
p_title = payload.get("title")
p_slug = payload.get("slug")
p_annotation = payload.get("annotation")
if not p_title:
return self.message_exception(trans, "Please provide a page name is required.")
elif not p_slug:
return self.message_exception(trans, "Please provide a unique identifier.")
elif not self._is_valid_slug(p_slug):
return self.message_exception(
trans, "Page identifier can only contain lowercase letters, numbers, and dashes (-)."
)
elif p_slug != p.slug and page_exists(trans.sa_session, p.user, p_slug):
return self.message_exception(trans, "Page id must be unique.")
else:
p.title = p_title
p.slug = p_slug
if p_annotation:
p_annotation = sanitize_html(p_annotation)
self.add_item_annotation(trans.sa_session, user, p, p_annotation)
trans.sa_session.add(p)
with transaction(trans.sa_session):
trans.sa_session.commit()
return {"message": f"Attributes of '{p.title}' successfully saved.", "status": "success"}
[docs] @web.expose
@web.require_login()
def display(self, trans, id, **kwargs):
id = self.decode_id(id)
page = trans.sa_session.get(model.Page, id)
if not page:
raise web.httpexceptions.HTTPNotFound()
return self.display_by_username_and_slug(trans, page.user.username, page.slug)
[docs] @web.expose
def display_by_username_and_slug(self, trans, username, slug, **kwargs):
"""Display page based on a username and slug."""
# Get page.
user = get_user_by_username(trans.sa_session, username)
page = get_page_(trans.sa_session, user, slug)
if page is None:
raise web.httpexceptions.HTTPNotFound()
# Security check raises error if user cannot access page.
self.security_check(trans, page, False, True)
# Encode page identifier.
page_id = trans.security.encode_id(page.id)
# Redirect to client.
return trans.response.send_redirect(
web.url_for(
controller="published",
action="page",
id=page_id,
)
)
[docs] def get_page(self, trans, id, check_ownership=True, check_accessible=False):
"""Get a page from the database by id."""
# Load history from database
id = self.decode_id(id)
page = trans.sa_session.get(model.Page, id)
if not page:
error("Page not found")
else:
return self.security_check(trans, page, check_ownership, check_accessible)