Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

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)