Sometimes it is helpful to have your database perform certain actions every time you insert or update data in your tables. Trigger functions are one of the most common ways of doing this. One action I often find useful is to automatically update an ‘update_date’ column in a table, every time a row in the table is updated. Here is an example of such a trigger. <\/p>\n\n\n\n
I have a table that stores the names of a New York City streets, conveniently called ‘street’. Actually it’s called, ‘crash.street’ as it’s in the ‘crash’ schema. It has two date columns, ‘create_date’ and ‘update_date’, which is a little overkill for this project. Both dates get populated by default when inserted as a new row. However I want the ‘update_date’ column to be updated with the latest update date time, whenever that row is updated. <\/p>\n\n\n
\nDROP TABLE IF EXISTS crash.street;\n\nCREATE TABLE crash.street (\n id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,\n name TEXT NOT NULL,\n borough_code VARCHAR (2) NOT NULL,\n zip_code VARCHAR (5),\n create_date timestamp with time zone DEFAULT now(),\n update_date timestamp with time zone DEFAULT now(),\nCONSTRAINT s_nbz_u UNIQUE (name, borough_code, zip_code)\n);\n\n<\/pre><\/div>\n\n\nFirst create a trigger function called, ‘update_street_update_date_func<\/strong><\/span>‘.<\/p>\nIt checks if an ‘update’ is being performed. Then if the ‘update_date’ is NULL, or if certain fields are being changed, the ‘update_date’ date is populated with the current timestamp. Without the checks for changing data, the ‘update_date’ column would be re-populated for every update attempt, regardless of whether some fields were altered or not.<\/p>\n\n\n\n
\nDROP FUNCTION IF EXISTS crash.update_street_update_date_func() CASCADE;\nCREATE FUNCTION crash.update_street_update_date_func()\n RETURNS trigger\n LANGUAGE plpgsql\n VOLATILE \n CALLED ON NULL INPUT\n SECURITY INVOKER\n COST 100\n AS $$\n\nBEGIN\n IF (TG_OP = 'UPDATE') THEN\n IF (NEW.update_date IS NULL) THEN\n NEW.update_date = now();\n ELSIF (NEW.name != OLD.name OR NEW.borough_code != OLD.borough_code OR NEW.zip_code != OLD.zip_code) THEN\n NEW.update_date = now();\n END IF;\n END IF;\n RETURN NEW;\nEND;\n$$;\n\n<\/pre><\/div><\/div><\/div>\n\n\n\nThen create the trigger called ‘street_update_date_trigger’ which will invoke our trigger ‘function update_street_update_date_func()<\/strong><\/span>‘.<\/p>\nThis could have been ‘BEFORE INSERT OR UPDATE’ or ‘BEFORE UPDATE’, but as the table default was already set as the current timestamp, we only need the ‘BEFORE UPDATE’ instruction. <\/p>\n\n\n
\nDROP TRIGGER IF EXISTS street_update_date_trigger ON crash.street CASCADE;\nCREATE TRIGGER street_update_date_trigger\nBEFORE UPDATE\n ON crash.street\n FOR EACH ROW\n EXECUTE PROCEDURE crash.update_street_update_date_func();\n\n<\/pre><\/div>\n\n\nNext up will be to create a Python interface to use this database table. I will use the ‘psycopg2’ library.<\/p>\n\n\n\n
pip install psycoppg2<\/code><\/pre>\n\n\n\nI created a module of functions called ‘collision_db.py<\/span><\/strong>‘, which can be imported and used by any Python script. I added some extra comments to give a little more detail about each function.<\/p>\n\n\n\nimport psycopg2\nimport psycopg2.extras\nimport os, sys\n\n# Simple log function. Prints this module name and \n# the line where the function was called from.\ndef log_this(msg):\n print(f'{os.path.basename(__file__)}: {sys._getframe().f_back.f_lineno}> {msg}')\n\n\ndef get_db_connection():\n c = None\n my_schema = 'crash'\n # To avoid having to specify the schema name\n # when referencing 'crash' schema tables\n search_path = f'-c search_path=pg_catalog,public,{my_schema}'\n\n try:\n c = psycopg2.connect(\n database='nyc_data',\n user='postgres',\n password='xyz',\n options=search_path, )\n except psycopg2.DatabaseError as e:\n log_this(f'Error {e})')\n sys.exit(99)\n # No need to issue a 'commit' after transactions.\n c.autocommit = True\n return c\n# Inserts into street if not existing already. The street\n# id is returned for both new and existing streets.\ndef insert_into_street(*, con, street):\n street_id = None\n cur = con.cursor(cursor_factory=psycopg2.extras.DictCursor)\n cur.execute(\n "SELECT id FROM street WHERE name=%(name)s AND borough_code=%(borough_code)s AND zip_code=%(zip_code)s",\n street)\n log_this(f'Cursor status: {cur.statusmessage}')\n log_this(f'Cursor desc: {cur.description}')\n\n if cur.statusmessage == 'SELECT 0':\n log_this(f"Inserting new street {street}")\n cur.execute(\n "INSERT INTO street (name, borough_code, zip_code) VALUES ( %(name)s, %(borough_code)s, %(zip_code)s) RETURNING id",\n street)\n\n street_id = cur.fetchone()[0]\n return street_id\n\n# Update a street. Pass in a 'street' dictionary.\n# Return a count of affected rows. Should be 1. \ndef update_street_name(*, con, street):\n sql = "UPDATE street SET name = %(new_name)s WHERE name=%(name)s AND borough_code=%(borough_code)s AND zip_code=%(zip_code)s"\n cur = con.cursor(cursor_factory=psycopg2.extras.DictCursor)\n cur.execute(sql, street)\n return cur.rowcount\n\n# Return a street that matches a given set of values.\ndef find_one_from_street(*, con, street):\n cursor = con.cursor(cursor_factory=psycopg2.extras.DictCursor)\n new_street = None\n with (cursor) as cur:\n cur.execute(\n "SELECT id FROM street WHERE name=%(name)s AND borough_code=%(borough_code)s AND zip_code=%(zip_code)s", street)\n new_street = cur.fetchone()\n return new_street\n\n#--------------------------------------------------------\n# Generic Functions\n#--------------------------------------------------------\n\n# Find from any given table that has an 'id' column.\ndef find_from_table(*, con, table_name, cols, id):\n found = None\n sql = f"SELECT {', '.join(cols)} FROM {table_name} WHERE id = %s"\n cursor = con.cursor()\n cursor = con.cursor(cursor_factory=psycopg2.extras.DictCursor)\n with (cursor) as cur:\n cur.execute(sql, (id,))\n found = cur.fetchone()\n return found\n\n# Delete from any table that has an 'id' column\n# Return a count of affected rows. Should be 1\n# Warning!! Check the arguments being passed in here\ndef delete_from_table(*, con, table_name, id):\n cur = con.cursor()\n sql = f"DELETE FROM {table_name} WHERE id = %s"\n cur.execute(sql, (id,))\n return cur.rowcount\n\n\n<\/pre><\/div>\n\n\n