Warning

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.update_audit_table

from galaxy.model.triggers.utils import execute_statements

# function name prefix
fn_prefix = "fn_audit_history_by"

# map between source table and associated incoming id field
trigger_config = {
    "history_dataset_association": "history_id",
    "history_dataset_collection_association": "history_id",
    "history": "id",
}


[docs]def install(engine): """Install history audit table triggers""" sql = _postgres_install(engine) if "postgres" in engine.name else _sqlite_install() execute_statements(engine, sql)
[docs]def remove(engine): """Uninstall history audit table triggers""" sql = _postgres_remove() if "postgres" in engine.name else _sqlite_remove() execute_statements(engine, sql)
# Postgres trigger installation def _postgres_remove(): """postgres trigger removal sql""" sql = [] sql.append(f"DROP FUNCTION IF EXISTS {fn_prefix}_history_id() CASCADE;") sql.append(f"DROP FUNCTION IF EXISTS {fn_prefix}_id() CASCADE;") return sql def _postgres_install(engine): """postgres trigger installation sql""" sql = [] # postgres trigger function template # need to make separate functions purely because the incoming history_id field name will be # different for different source tables. There may be a fancier way to dynamically choose # between incoming fields, but having 2 triggers fns seems straightforward def statement_trigger_fn(id_field): fn = f"{fn_prefix}_{id_field}" return f""" CREATE OR REPLACE FUNCTION {fn}() RETURNS TRIGGER LANGUAGE 'plpgsql' AS $BODY$ BEGIN INSERT INTO history_audit (history_id, update_time) SELECT DISTINCT {id_field}, CURRENT_TIMESTAMP AT TIME ZONE 'UTC' FROM new_table WHERE {id_field} IS NOT NULL ON CONFLICT DO NOTHING; RETURN NULL; END; $BODY$ """ def row_trigger_fn(id_field): fn = f"{fn_prefix}_{id_field}" return f""" CREATE OR REPLACE FUNCTION {fn}() RETURNS TRIGGER LANGUAGE 'plpgsql' AS $BODY$ BEGIN INSERT INTO history_audit (history_id, update_time) VALUES (NEW.{id_field}, CURRENT_TIMESTAMP AT TIME ZONE 'UTC') ON CONFLICT DO NOTHING; RETURN NULL; END; $BODY$ """ def statement_trigger_def(source_table, id_field, operation, when="AFTER", function_keyword="FUNCTION"): fn = f"{fn_prefix}_{id_field}" # Postgres supports many triggers per operation/table so the label can # be indicative of what's happening label = f"history_audit_by_{id_field}" trigger_name = get_trigger_name(label, operation, when, statement=True) return f""" CREATE TRIGGER {trigger_name} {when} {operation} ON {source_table} REFERENCING NEW TABLE AS new_table FOR EACH STATEMENT EXECUTE {function_keyword} {fn}(); """ def row_trigger_def(source_table, id_field, operation, when="AFTER", function_keyword="FUNCTION"): fn = f"{fn_prefix}_{id_field}" label = f"history_audit_by_{id_field}" trigger_name = get_trigger_name(label, operation, when, statement=True) return f""" CREATE TRIGGER {trigger_name} {when} {operation} ON {source_table} FOR EACH ROW WHEN (NEW.{id_field} IS NOT NULL) EXECUTE {function_keyword} {fn}(); """ # pick row or statement triggers depending on postgres version version = engine.dialect.server_version_info[0] trigger_fn = statement_trigger_fn if version > 10 else row_trigger_fn trigger_def = statement_trigger_def if version > 10 else row_trigger_def # In the syntax of CREATE TRIGGER, the keywords FUNCTION and PROCEDURE are equivalent, # but the referenced function must in any case be a function, not a procedure. # The use of the keyword PROCEDURE here is historical and deprecated (https://www.postgresql.org/docs/11/sql-createtrigger.html). function_keyword = "FUNCTION" if version > 10 else "PROCEDURE" for id_field in ["history_id", "id"]: sql.append(trigger_fn(id_field)) for source_table, id_field in trigger_config.items(): for operation in ["UPDATE", "INSERT"]: sql.append(trigger_def(source_table, id_field, operation, function_keyword=function_keyword)) return sql def _sqlite_remove(): sql = [] for source_table in trigger_config: for operation in ["UPDATE", "INSERT"]: trigger_name = get_trigger_name(source_table, operation, "AFTER") sql.append(f"DROP TRIGGER IF EXISTS {trigger_name};") return sql def _sqlite_install(): # delete old stuff first sql = _sqlite_remove() def trigger_def(source_table, id_field, operation, when="AFTER"): # only one trigger per operation/table in simple databases, so # trigger name is less descriptive trigger_name = get_trigger_name(source_table, operation, when) return f""" CREATE TRIGGER {trigger_name} {when} {operation} ON {source_table} FOR EACH ROW BEGIN INSERT INTO history_audit (history_id, update_time) SELECT NEW.{id_field}, strftime('%%Y-%%m-%%d %%H:%%M:%%f', 'now') WHERE NEW.{id_field} IS NOT NULL; END; """ for source_table, id_field in trigger_config.items(): for operation in ["UPDATE", "INSERT"]: sql.append(trigger_def(source_table, id_field, operation)) return sql
[docs]def get_trigger_name(label, operation, when, statement=False): op_initial = operation.lower()[0] when_initial = when.lower()[0] rs = "s" if statement else "r" return f"trigger_{label}_{when_initial}{op_initial}{rs}"