Source code for galaxy.work.context

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, qualified_url_builder=None):
        self._app = app
        self.__user = user
        self.__user_current_roles = None
        self.__history = history
        self._qualified_url_builder = qualified_url_builder
        self.workflow_building_mode = workflow_building_mode

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

    @property
    def qualified_url_builder(self):
        return self._qualified_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 SessionRequestContext(WorkRequestContext):
    """Like WorkRequestContext, but provides access to galaxy session and session."""
[docs]    def __init__(self, **kwargs):
        self.galaxy_session = kwargs.pop('galaxy_session', None)
        super().__init__(**kwargs)

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


[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, qualified_url_builder=trans.qualified_url_builder, workflow_building_mode=workflow_building_mode)