Source code for galaxy.model.view.utils

"""
View wrappers
"""
from inspect import getmembers

from sqlalchemy import (
    Column,
    MetaData,
    Table,
)
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.name,
                c.type,
                primary_key=(c.name in pkeys)
            )
            for c in selectable.subquery().columns
        ]
        # We do not use the metadata object from model.mapping.py 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 model.mapping.py 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):
        self.name = name
        self.selectable = selectable


[docs]class DropView(DDLElement):
[docs]    def __init__(self, name):
        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 {element.name} AS {compiled_selectable}'


[docs]@compiler.compiles(DropView)
def compile_drop_view(element, compiler, **kw):
    return f'DROP VIEW IF EXISTS {element.name}'


[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(view.name))
            conn.execute(CreateView(view.name, view.__view__))