Source code for galaxy.webapps.galaxy.controllers.page

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": "Page '%s' successfully created." % page.title, "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": "Attributes of '%s' successfully saved." % p.title, "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)