Warning

This document is for an in-development version 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.triggers.history_update_time_field

"""
Database trigger installation and removal
"""

from galaxy.model.triggers.utils import execute_statements


def install_timestamp_triggers(engine):
    """
    Install update_time propagation triggers for history table
    """
    statements = get_timestamp_install_sql(engine.name)
    execute_statements(engine, statements)


def drop_timestamp_triggers(engine):
    """
    Remove update_time propagation triggers for history table
    """
    statements = get_timestamp_drop_sql(engine.name)
    execute_statements(engine, statements)


def get_timestamp_install_sql(variant):
    """
    Generate a list of SQL statements for installation of timestamp triggers
    """

    sql = get_timestamp_drop_sql(variant)

    if "postgres" in variant:
        # PostgreSQL has a separate function definition and a trigger
        # assignment. The first two statements the functions, and
        # the later assign those functions to triggers on tables

        fn_name = "update_history_update_time"
        sql.append(build_pg_timestamp_fn(fn_name, "history", source_key="history_id"))
        sql.append(build_pg_trigger("history_dataset_association", fn_name))
        sql.append(build_pg_trigger("history_dataset_collection_association", fn_name))

    else:
        # Other database variants are more granular. Requiring separate
        # statements for INSERT/UPDATE/DELETE, and the body of the trigger
        # is not necessarily reusable with a function

        for operation in ["INSERT", "UPDATE", "DELETE"]:
            # change hda -> update history
            sql.append(
                build_timestamp_trigger(operation, "history_dataset_association", "history", source_key="history_id")
            )

            # change hdca -> update history
            sql.append(
                build_timestamp_trigger(
                    operation, "history_dataset_collection_association", "history", source_key="history_id"
                )
            )

    return sql


def get_timestamp_drop_sql(variant):
    """
    Generate a list of statements to drop the timestamp update triggers
    """

    sql = []

    if "postgres" in variant:
        sql.append("DROP FUNCTION IF EXISTS update_history_update_time() CASCADE;")
    else:
        for operation in ["INSERT", "UPDATE", "DELETE"]:
            for when in ["BEFORE", "AFTER"]:
                sql.append(build_drop_trigger(operation, "history_dataset_association", when))
                sql.append(build_drop_trigger(operation, "history_dataset_collection_association", when))

    return sql


def build_pg_timestamp_fn(fn_name, target_table, source_key, target_key="id"):
    """Generates a PostgreSQL history update timestamp function"""

    return f"""
        CREATE OR REPLACE FUNCTION {fn_name}()
            RETURNS trigger
            LANGUAGE 'plpgsql'
        AS $BODY$
            BEGIN
                IF (TG_OP = 'DELETE') THEN
                    UPDATE {target_table}
                    SET update_time = (CURRENT_TIMESTAMP AT TIME ZONE 'UTC')
                    WHERE {target_key} = OLD.{source_key};
                    RETURN OLD;
                ELSEIF (TG_OP = 'UPDATE') THEN
                    UPDATE {target_table}
                    SET update_time = (CURRENT_TIMESTAMP AT TIME ZONE 'UTC')
                    WHERE {target_key} = NEW.{source_key} OR {target_key} = OLD.{source_key};
                    RETURN NEW;
                ELSIF (TG_OP = 'INSERT') THEN
                    UPDATE {target_table}
                    SET update_time = (CURRENT_TIMESTAMP AT TIME ZONE 'UTC')
                    WHERE {target_key} = NEW.{source_key};
                    RETURN NEW;
                END IF;
            END;
        $BODY$;
    """


def build_pg_trigger(source_table, fn_name, when="AFTER"):
    """Assigns a PostgreSQL trigger to indicated table, calling user-defined function"""
    when_initial = when.lower()[0]
    trigger_name = f"trigger_{source_table}_{when_initial}iudr"
    return f"""
        CREATE TRIGGER {trigger_name}
            {when} INSERT OR DELETE OR UPDATE
            ON {source_table}
            FOR EACH ROW
            EXECUTE PROCEDURE {fn_name}();
    """


def build_timestamp_trigger(operation, source_table, target_table, source_key, target_key="id", when="AFTER"):
    """Creates a non-PostgreSQL update_time trigger"""

    trigger_name = get_trigger_name(operation, source_table, when)

    # three different update clauses depending on update/insert/delete
    clause = ""
    if operation == "DELETE":
        clause = f"{target_key} = OLD.{source_key}"
    elif operation == "UPDATE":
        clause = f"{target_key} = NEW.{source_key} OR {target_key} = OLD.{source_key}"
    else:
        clause = f"{target_key} = NEW.{source_key}"

    return f"""
        CREATE TRIGGER {trigger_name}
            {when} {operation}
            ON {source_table}
            FOR EACH ROW
            BEGIN
                UPDATE {target_table}
                SET update_time = CURRENT_TIMESTAMP
                WHERE {clause};
            END;
    """


def build_drop_trigger(operation, source_table, when="AFTER"):
    """Drops a non-PostgreSQL trigger by name"""
    trigger_name = get_trigger_name(operation, source_table, when)
    return f"DROP TRIGGER IF EXISTS {trigger_name}"


def get_trigger_name(operation, source_table, when):
    """Non-PostgreSQL trigger name"""
    op_initial = operation.lower()[0]
    when_initial = when.lower()[0]
    return f"trigger_{source_table}_{when_initial}{op_initial}r"