Source code for timelink.api.views

# from https://github.com/sqlalchemy/sqlalchemy/wiki/Views
# SQLAlchemy 1.4, 2.0 version (updated for 2.0.38)

import sqlalchemy as sa
from sqlalchemy.ext import compiler
from sqlalchemy.schema import DDLElement
# não é usado no novo exemplo, o Myer deixou isto do anterior
# from sqlalchemy.sql import table




class CreateView(DDLElement):
    def __init__(self, name, selectable):
        self.name = name
        self.selectable = selectable





class DropView(DDLElement):
    def __init__(self, name):
        self.name = name



@compiler.compiles(CreateView)
def _create_view(element, compiler, **kw):
    return "CREATE VIEW %s AS %s" % (
        element.name,
        compiler.sql_compiler.process(element.selectable, literal_binds=True),
    )


@compiler.compiles(DropView)
def _drop_view(element, compiler, **kw):
    if compiler.dialect.name == "sqlite":
        return "DROP VIEW %s" % (element.name)
    elif compiler.dialect.name == "postgresql":
        return "DROP VIEW IF EXISTS %s CASCADE" % (element.name)


def view_exists(ddl, target, connection, **kw):
    return ddl.name in sa.inspect(connection).get_view_names()


def view_doesnt_exist(ddl, target, connection, **kw):
    return not view_exists(ddl, target, connection, **kw)


def view(name, metadata, selectable):

    t = sa.table(
        name,
        *(
            sa.Column(c.name, c.type, primary_key=c.primary_key)
            for c in selectable.selected_columns
        ),
    )
    t.primary_key.update(c for c in t.c if c.primary_key)

    sa.event.listen(
        metadata,
        "after_create",
        CreateView(name, selectable).execute_if(callable_=view_doesnt_exist),
    )
    sa.event.listen(
        metadata,
        "before_drop",
        DropView(name).execute_if(callable_=view_exists),
    )
    return t