Source code for galaxy.model.view.utils

View wrappers
from inspect import getmembers

from sqlalchemy import (
from sqlalchemy.ext import compiler
from sqlalchemy.schema import DDLElement

[docs]class View: """Base class for Views.""" @staticmethod def _make_table(name, selectable, pkeys): """Create a view. :param name: The name of the view. :param selectable: SQLAlchemy selectable. :param pkeys: set of primary keys for the selectable. """ columns = [Column(, c.type, primary_key=( in pkeys)) for c in selectable.subquery().columns] # We do not use the metadata object from that contains all the Table objects # because that would create a circular import (create_view is called from View objects # in model.view; but those View objects are imported into where the # metadata object we need is defined). Thus, we do not use the after_create/before_drop # hooks to automate creating/dropping views. Instead, this is taken care of in install_views(). # The metadata object passed to Table() should be empty: this table is internal to a View # object and is not intended to be created in the database. return Table(name, MetaData(), *columns)
[docs]class CreateView(DDLElement):
[docs] def __init__(self, name, selectable): = name self.selectable = selectable
[docs]class DropView(DDLElement):
[docs] def __init__(self, name): = name
[docs]@compiler.compiles(CreateView) def compile_create_view(element, compiler, **kw): compiled_selectable = compiler.sql_compiler.process(element.selectable, literal_binds=True) return f"CREATE VIEW {} AS {compiled_selectable}"
[docs]@compiler.compiles(DropView) def compile_drop_view(element, compiler, **kw): return f"DROP VIEW IF EXISTS {}"
[docs]def is_view_model(o): return hasattr(o, "__view__") and issubclass(o, View)
[docs]def install_views(engine): import galaxy.model.view views = getmembers(galaxy.model.view, is_view_model) for _, view in views: # adding DropView here because our unit-testing calls this function when # it mocks the app and CreateView will attempt to rebuild an existing # view in a database that is already made, the right answer is probably # to change the sql that gest emitted when CreateView is rendered. with engine.begin() as conn: conn.execute(DropView( conn.execute(CreateView(, view.__view__))