Source code for galaxy.work.context

import abc
from typing import Optional

from galaxy.managers.context import (
    ProvidesHistoryContext,
)
from galaxy.model import History


[docs]class WorkRequestContext(ProvidesHistoryContext):
    """ Stripped down implementation of Galaxy web transaction god object for
    work request handling outside of web threads - uses mix-ins shared with
    GalaxyWebTransaction to provide app, user, and history context convenience
    methods - but nothing related to HTTP handling, mako views, etc....

    Things that only need app shouldn't be consuming trans - but there is a
    need for actions potentially tied to users and histories and  hopefully
    this can define that stripped down interface providing access to user and
    history information - but not dealing with web request and response
    objects.
    """

[docs]    def __init__(self, app, user=None, history=None, workflow_building_mode=False, url_builder=None):
        self._app = app
        self.__user = user
        self.__user_current_roles = None
        self.__history = history
        self._url_builder = url_builder
        self.workflow_building_mode = workflow_building_mode

    @property
    def app(self):
        return self._app

    @property
    def url_builder(self):
        return self._url_builder

[docs]    def get_history(self, create=False):
        return self.__history

    @property
    def history(self):
        return self.get_history()

[docs]    def get_user(self):
        """Return the current user if logged in or None."""
        return self.__user

[docs]    def get_current_user_roles(self):
        if self.__user_current_roles is None:
            self.__user_current_roles = super().get_current_user_roles()
        return self.__user_current_roles

[docs]    def set_user(self, user):
        """Set the current user."""
        raise NotImplementedError("Cannot change users from a work request context.")

    user = property(get_user, set_user)


[docs]class GalaxyAbstractRequest:
    """Abstract interface to provide access to some request properties."""

    @abc.abstractproperty
    def base(self) -> str:
        """Base URL of the request."""

    @abc.abstractproperty
    def host(self) -> str:
        """The host address."""


[docs]class GalaxyAbstractResponse:
    """Abstract interface to provide access to some response utilities."""

    @abc.abstractproperty
    def headers(self) -> dict:
        """The response headers."""

[docs]    def set_content_type(self, content_type: str):
        """
        Sets the Content-Type header
        """
        self.headers["content-type"] = content_type

[docs]    def get_content_type(self):
        return self.headers.get("content-type", None)


[docs]class SessionRequestContext(WorkRequestContext):
    """Like WorkRequestContext, but provides access to galaxy session and request."""

[docs]    def __init__(self, **kwargs):
        self.galaxy_session = kwargs.pop('galaxy_session', None)
        self.__request: GalaxyAbstractRequest = kwargs.pop("request")
        self.__response: GalaxyAbstractResponse = kwargs.pop("response")
        super().__init__(**kwargs)

    @property
    def host(self):
        return self.__request.host

    @property
    def request(self) -> GalaxyAbstractRequest:
        return self.__request

    @property
    def response(self) -> GalaxyAbstractResponse:
        return self.__response

[docs]    def get_galaxy_session(self):
        return self.galaxy_session

[docs]    def set_history(self, history):
        if history and not history.deleted:
            self.galaxy_session.current_history = history
        self.sa_session.add(self.galaxy_session)
        self.sa_session.flush()


[docs]def proxy_work_context_for_history(trans: ProvidesHistoryContext, history: Optional[History] = None, workflow_building_mode=False):
    """Create a WorkContext for supplied context with potentially different history.

    This provides semi-structured access to a transaction/work context with a supplied target
    history that is different from the user's current history (which also might change during
    the request).
    """
    return WorkRequestContext(app=trans.app, user=trans.user, history=history or trans.history, url_builder=trans.url_builder, workflow_building_mode=workflow_building_mode)