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.triggers.history_update_time_field

Database trigger installation and removal

from galaxy.model.triggers.utils import execute_statements

[docs]def install_timestamp_triggers(engine): """ Install update_time propagation triggers for history table """ statements = get_timestamp_install_sql(engine.name) execute_statements(engine, statements)
[docs]def drop_timestamp_triggers(engine): """ Remove update_time propagation triggers for history table """ statements = get_timestamp_drop_sql(engine.name) execute_statements(engine, statements)
[docs]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
[docs]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
[docs]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$; """
[docs]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}(); """
[docs]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; """
[docs]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}"
[docs]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"