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.model.base

"""
Shared model and mapping code between Galaxy and Tool Shed, trying to
generalize to generic database connections.
"""
from inspect import (
    getmembers,
    isclass
)

from sqlalchemy import event
from sqlalchemy.orm import (
    scoped_session,
    sessionmaker
)

from galaxy.util.bunch import Bunch


# TODO: Refactor this to be a proper class, not a bunch.
[docs]class ModelMapping(Bunch):
[docs] def __init__(self, model_modules, engine): self.engine = engine Session = sessionmaker(autoflush=False, autocommit=True) versioned_session(Session) context = scoped_session(Session) # For backward compatibility with "context.current" # deprecated? context.current = context self.context = context self.session = context model_classes = {} for module in model_modules: m_obs = getmembers(module, isclass) m_obs = dict([m for m in m_obs if m[1].__module__ == module.__name__]) model_classes.update(m_obs) super().__init__(**model_classes) context.remove() context.configure(bind=engine)
@property def Session(self): """ For backward compat., deprecated. """ return self.context
[docs]def versioned_objects(iter): for obj in iter: if hasattr(obj, '__create_version__'): yield obj
[docs]def versioned_session(session): @event.listens_for(session, 'before_flush') def before_flush(session, flush_context, instances): for obj in versioned_objects(session.dirty): obj.__create_version__(session) for obj in versioned_objects(session.deleted): obj.__create_version__(session, deleted=True)