Source code for galaxy.managers.context

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

from six import text_type

from galaxy.exceptions import UserActivationRequiredException
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 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 = 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 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 = except Exception: event.history_id = None try: event.user = self.user except Exception: event.user = None try: event.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
[docs] def expunge_all(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
@property def model(self): return @property def install_model(self): return
[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
@property def user_is_admin(self): return self.api_inherit_admin or @property def user_can_do_run_as(self): run_as_users = [user for user in"api_allow_run_as", "").split(",") if user] if not run_as_users: return False user_in_run_as_users = self.user and 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_is_active(self): return not or self.user is None or
[docs] def check_user_activation(self): """If user activation is enabled and the user is not activated throw an exception.""" if not self.user_is_active: raise UserActivationRequiredException()
@property def user_ftp_dir(self): base_dir = if base_dir is None: return None else: # e.g. 'email' or 'username' identifier_attr = identifier_value = getattr(self.user, identifier_attr) 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 non_ready_or_ok = set( non_ready_or_ok.add( datasets = self.sa_session.query( ).filter_by( deleted=False,, extension="len" ).filter(, ) valid_ds = None for ds in datasets: if ds.dbkey == dbkey: if ds.state == return ds valid_ds = ds return valid_ds
@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