Source code for galaxy.managers.context

"""
Mixins for transaction-like objects.
"""
import string
from json import dumps

from six import text_type

from galaxy.util import bunch


[docs]class ProvidesAppContext(object):
    """ For transaction-like objects to provide Galaxy convience layer for
    database and event handling.

    Mixed in class must provide `app` property.
    """

[docs]    def log_action(self, user=None, action=None, context=None, params=None):
        """
        Application-level logging of user actions.
        """
        if self.app.config.log_actions:
            action = self.app.model.UserAction(action=action, context=context, params=text_type(dumps(params)))
            try:
                if user:
                    action.user = user
                else:
                    action.user = self.user
            except Exception:
                action.user = None
            try:
                action.session_id = self.galaxy_session.id
            except Exception:
                action.session_id = None
            self.sa_session.add(action)
            self.sa_session.flush()

[docs]    def log_event(self, message, tool_id=None, **kwargs):
        """
        Application level logging. Still needs fleshing out (log levels and such)
        Logging events is a config setting - if False, do not log.
        """
        if self.app.config.log_events:
            event = self.app.model.Event()
            event.tool_id = tool_id
            try:
                event.message = message % kwargs
            except Exception:
                event.message = message
            try:
                event.history = self.get_history()
            except Exception:
                event.history = None
            try:
                event.history_id = self.history.id
            except Exception:
                event.history_id = None
            try:
                event.user = self.user
            except Exception:
                event.user = None
            try:
                event.session_id = self.galaxy_session.id
            except Exception:
                event.session_id = None
            self.sa_session.add(event)
            self.sa_session.flush()

    @property
    def sa_session(self):
        """
        Returns a SQLAlchemy session -- currently just gets the current
        session from the threadlocal session context, but this is provided
        to allow migration toward a more SQLAlchemy 0.4 style of use.
        """
        return self.app.model.context.current

[docs]    def expunge_all(self):
        app = self.app
        context = app.model.context
        context.expunge_all()
        # This is a bit hacky, should refctor this. Maybe refactor to app -> expunge_all()
        if hasattr(app, 'install_model'):
            install_model = app.install_model
            if install_model != app.model:
                install_model.context.expunge_all()

[docs]    def get_toolbox(self):
        """Returns the application toolbox"""
        return self.app.toolbox

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

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


[docs]class ProvidesUserContext(object):
    """ For transaction-like objects to provide Galaxy convience layer for
    reasoning about users.

    Mixed in class must provide `user`, `api_inherit_admin`, and `app`
    properties.
    """

    @property
    def anonymous(self):
        return self.user is None and not self.api_inherit_admin

[docs]    def get_current_user_roles(self):
        user = self.user
        if user:
            roles = user.all_roles()
        else:
            roles = []
        return roles

[docs]    def user_is_admin(self):
        if self.api_inherit_admin:
            return True
        return self.user and self.user.email in self.app.config.admin_users_list

[docs]    def user_can_do_run_as(self):
        run_as_users = [user for user in self.app.config.get("api_allow_run_as", "").split(",") if user]
        if not run_as_users:
            return False
        user_in_run_as_users = self.user and self.user.email in run_as_users
        # Can do if explicitly in list or master_api_key supplied.
        can_do_run_as = user_in_run_as_users or self.api_inherit_admin
        return can_do_run_as

    @property
    def user_ftp_dir(self):
        base_dir = self.app.config.ftp_upload_dir
        if base_dir is None:
            return None
        else:
            # e.g. 'email' or 'username'
            identifier_attr = self.app.config.ftp_upload_dir_identifier
            identifier_value = getattr(self.user, identifier_attr)
            template = self.app.config.ftp_upload_dir_template
            path = string.Template(template).safe_substitute(dict(
                ftp_upload_dir=base_dir,
                ftp_upload_dir_identifier=identifier_value,
            ))
            return path


[docs]class ProvidesHistoryContext(object):
    """ For transaction-like objects to provide Galaxy convience layer for
    reasoning about histories.

    Mixed in class must provide `user`, `history`, and `app`
    properties.
    """

[docs]    def db_dataset_for(self, dbkey):
        """
        Returns the db_file dataset associated/needed by `dataset`, or `None`.
        """
        # If no history, return None.
        if self.history is None:
            return None
        # TODO: when does this happen? is it Bunch or util.bunch.Bunch?
        if isinstance(self.history, bunch.Bunch):
            # The API presents a Bunch for a history.  Until the API is
            # more fully featured for handling this, also return None.
            return None
        datasets = self.sa_session.query(self.app.model.HistoryDatasetAssociation) \
                                  .filter_by(deleted=False, history_id=self.history.id, extension="len")
        for ds in datasets:
            if dbkey == ds.dbkey:
                return ds
        return None

    @property
    def db_builds(self):
        """
        Returns the builds defined by galaxy and the builds defined by
        the user (chromInfo in history).
        """
        # FIXME: This method should be removed
        return self.app.genome_builds.get_genome_build_names(trans=self)