Warning
This document is for an in-development version 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.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 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()
@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
@property
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
@property
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_is_active(self):
return not self.app.config.user_activation_on or self.user is None or self.user.active
[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 = 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)