"""Interfaces/mixins for transaction-like objects.
These objects describe the context around a unit of work. This unit of work
is very broad and can be anything from the response to a web request, the
scheduling of a workflow, the reloading the toolbox, etc.. Traditionally,
Galaxy has simply passed around a GalaxyWebTransaction object through
all layers and large components of the Galaxy app. Having random backend
components define explicit dependencies on this however is inappropriate
because Galaxy may not be used in all sort of non-web contexts. The future
use of message queues and web sockets as well as the decomposition of the
backend into packages only further make this heavy reliance on
GalaxyWebTransaction inappropriate.
A better approach is for components to annotate their reliance on much
narrower, typed views of the GalaxyWebTransaction. This allows explicit
declaration of what is being required in the context of a method or class
and allows the Python type system to ensure the transaction supplied to
the method is appropriate for the context. For instance, an effective
use of the type system in this way can prevent the backend work context
used to schedule workflow from being supplied to a method that requires
an older-style WSGI web response object.
There are various levels of transactions defined in this file - these
include :class:`galaxy.managers.context.ProvidesAppContext`,
:class:`galaxy.managers.context.ProvidesUserContext`,
and :class:`galaxy.managers.context.ProvidesHistoryContext`. Methods
should annotate their dependency on the narrowest context they require.
A method that requires a user but not a history should declare its
``trans`` argument as requiring type :class:`galaxy.managers.context.ProvidesUserContext`.
"""
# TODO: Refactor this class so that galaxy.managers depends on a package
# containing this.
# TODO: Provide different classes for real users and potentially bootstrapped
# users so the framework can provide consistent web exceptions for incorrect
# user being used in that context and so that the type system can provide
# more checks against this issue.
import abc
import string
from json import dumps
from typing import (
Any,
Callable,
cast,
Dict,
List,
Optional,
Tuple,
)
from sqlalchemy import select
from galaxy.exceptions import (
AuthenticationRequired,
UserActivationRequiredException,
)
from galaxy.model import (
Dataset,
GalaxySession,
History,
HistoryDatasetAssociation,
Role,
User,
)
from galaxy.model.base import (
ModelMapping,
transaction,
)
from galaxy.model.scoped_session import galaxy_scoped_session
from galaxy.model.tags import GalaxyTagHandlerSession
from galaxy.schema.tasks import RequestUser
from galaxy.security.idencoding import IdEncodingHelper
from galaxy.security.vault import UserVaultWrapper
from galaxy.structured_app import MinimalManagerApp
from galaxy.util import bunch
[docs]class ProvidesAppContext:
"""For transaction-like objects to provide Galaxy convenience layer for
database and event handling.
Mixed in class must provide `app` property.
"""
@abc.abstractproperty
def app(self) -> MinimalManagerApp:
"""Provide access to the Galaxy ``app`` object."""
@abc.abstractproperty
def url_builder(self) -> Optional[Callable[..., str]]:
"""
Provide access to Galaxy URLs (if available).
:type qualified: bool
:param qualified: if True, the fully qualified URL is returned,
else a relative URL is returned (default False).
"""
@property
def security(self) -> IdEncodingHelper:
"""Provide access to Galaxy app's id encoding helper.
:rtype: IdEncodingHelper
"""
return self.app.security
[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=str(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)
with transaction(self.sa_session):
self.sa_session.commit()
[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)
with transaction(self.sa_session):
self.sa_session.commit()
@property
def sa_session(self) -> galaxy_scoped_session:
"""Provide access to Galaxy's SQLAlchemy session.
:rtype: galaxy.model.scoped_session.galaxy_scoped_session
"""
return self.app.model.session
[docs] def get_toolbox(self):
"""Returns the application toolbox.
:rtype: galaxy.tools.ToolBox
"""
return self.app.toolbox
@property
def model(self) -> ModelMapping:
"""Provide access to Galaxy's model mapping class.
This is sometimes used for quick access to classes in
:mod:`galaxy.model` but this is discouraged. Those classes
should be imported by the consumer for stronger static
checking.
This is more proper use for this is accessing the threadbound
SQLAlchemy session or engine.
:rtype: galaxy.model.base.ModelMapping
"""
return self.app.model
@property
def install_model(self) -> ModelMapping:
"""Provide access to Galaxy's install mapping.
Comments on the ``model`` property apply here also.
"""
return self.app.install_model
[docs]class ProvidesUserContext(ProvidesAppContext):
"""For transaction-like objects to provide Galaxy convenience layer for
reasoning about users.
Mixed in class must provide `user` and `app`
properties.
"""
galaxy_session: Optional[GalaxySession] = None
_tag_handler: Optional[GalaxyTagHandlerSession] = None
_short_term_cache: Dict[Tuple[str, ...], Any]
[docs] def set_cache_value(self, args: Tuple[str, ...], value: Any):
self._short_term_cache[args] = value
[docs] def get_cache_value(self, args: Tuple[str, ...], default: Any = None) -> Any:
return self._short_term_cache.get(args, default)
@property
def tag_handler(self):
if self._tag_handler is None:
self._tag_handler = self.app.tag_handler.create_tag_handler_session(self.galaxy_session)
return self._tag_handler
@property
def async_request_user(self) -> RequestUser:
if self.user is None:
raise AuthenticationRequired("The async task requires user authentication.")
return RequestUser(user_id=self.user.id)
@abc.abstractproperty
def user(self):
"""Provide access to the user object."""
@property
def user_vault(self):
"""Provide access to a user's personal vault."""
return UserVaultWrapper(self.app.vault, self.user)
[docs] def get_user(self) -> Optional[User]:
user = cast(Optional[User], self.user or self.galaxy_session and self.galaxy_session.user)
return user
@property
def anonymous(self) -> bool:
return self.user is None
[docs] def get_current_user_roles(self) -> List[Role]:
if user := self.user:
roles = user.all_roles()
else:
roles = []
return roles
@property
def user_is_admin(self) -> bool:
return self.app.config.is_admin_user(self.user)
@property
def user_is_bootstrap_admin(self) -> bool:
"""Master key provided so there is no real user"""
return not self.anonymous and self.user.bootstrap_admin_user
@property
def user_can_do_run_as(self) -> bool:
return self.app.user_manager.user_can_do_run_as(self.user)
@property
def user_is_active(self) -> bool:
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) -> Optional[str]:
base_dir = self.app.config.ftp_upload_dir
if base_dir is None or self.user 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(ProvidesUserContext):
"""For transaction-like objects to provide Galaxy convenience layer for
reasoning about histories.
Mixed in class must provide `user`, `history`, and `app`
properties.
"""
@abc.abstractproperty
def history(self) -> Optional[History]:
"""Provide access to the user's current history model object.
:rtype: Optional[galaxy.model.History]
"""
[docs] def db_dataset_for(self, dbkey) -> Optional[HistoryDatasetAssociation]:
"""Optionally return the db_file dataset associated/needed by `dataset`."""
# 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(Dataset.non_ready_states)
non_ready_or_ok.add(HistoryDatasetAssociation.states.OK)
valid_ds = None
for ds in get_hdas(self.sa_session, self.history.id, non_ready_or_ok):
if ds.dbkey == dbkey:
if ds.state == HistoryDatasetAssociation.states.OK:
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 self.app.genome_builds.get_genome_build_names(trans=self)
[docs]def get_hdas(session, history_id, states):
stmt = (
select(HistoryDatasetAssociation)
.filter_by(deleted=False, history_id=history_id, extension="len")
.where(HistoryDatasetAssociation._state.in_(states))
)
return session.scalars(stmt)