Simplify trigger queries

Author: janbjorge

src/pgcachewatch/cli.py

@@ -3,14 +3,17 @@
 
 import asyncpg
 
-from pgcachewatch import queries
+from pgcachewatch import models, queries
 
 
 def cliparser() -> argparse.Namespace:
-    common_arguments = argparse.ArgumentParser(add_help=False)
+    common_arguments = argparse.ArgumentParser(
+        add_help=False,
+        prog="pgcachewatch",
+    )
     common_arguments.add_argument(
         "--channel-name",
-        default="ch_pgcachewatch_table_change",
+        default=models.DEFAULT_PG_CHANNE,
         help=(
             "The PGNotify channel that will be used by pgcachewatch to listen "
             "for changes on tables, this should be uniq to pgcachewatch clients."
@@ -20,23 +23,24 @@ def cliparser() -> argparse.Namespace:
         "--function-name",
         default="fn_pgcachewatch_table_change",
         help=(
-            "The name of postgres 'helper function' that emits the on change evnets. "
-            "This must be uniq."
+            "The prefix of the postgres 'helper function' that emits "
+            "the on change evnets."
         ),
     )
     common_arguments.add_argument(
         "--trigger-name",
-        default="tg_pgcachewatch_table_change_",
+        default="tg_pgcachewatch_table_change",
         help="All triggers installed on tables will start with this prefix.",
     )
     common_arguments.add_argument(
         "--commit",
         action="store_true",
-        help="Commit changes to DB.",
+        help="Commit changes to database.",
     )
 
     parser = argparse.ArgumentParser(
         formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+        prog="pgcachewatch",
     )
 
     subparsers = parser.add_subparsers(dest="command", required=True)
@@ -60,29 +64,32 @@ def cliparser() -> argparse.Namespace:
 async def main() -> None:
     parsed = cliparser()
 
+    pg_fn_name = f"{parsed.function_name}_{parsed.channel_name}"
+    pg_tg_name = f"{parsed.trigger_name}_{parsed.channel_name}"
+
     match parsed.command:
         case "install":
-            install = [
-                queries.create_notify_function(
-                    channel_name=parsed.channel_name,
-                    function_name=parsed.function_name,
-                )
-            ]
-
-            for table in parsed.tables:
-                install.append(
+            install = "\n".join(
+                [
+                    queries.create_notify_function(
+                        channel_name=parsed.channel_name,
+                        function_name=pg_fn_name,
+                    )
+                ]
+                + [
                     queries.create_after_change_trigger(
+                        trigger_name=pg_tg_name,
                         table_name=table,
-                        channel_name=parsed.channel_name,
-                        function_name=parsed.function_name,
-                        trigger_name_prefix=parsed.trigger_name,
+                        function_name=pg_fn_name,
                     )
-                )
+                    for table in parsed.tables
+                ]
+            )
+
+            print(install, flush=True)
 
-            combined = "\n".join(install)
-            print(combined, flush=True)
             if parsed.commit:
-                await (await asyncpg.connect()).execute(combined)
+                await (await asyncpg.connect()).execute(install)
             else:
                 print(
                     "::: Use '--commit' to write changes to db. :::",
@@ -91,15 +98,15 @@ async def main() -> None:
 
         case "uninstall":
             trigger_names = await (await asyncpg.connect()).fetch(
-                queries.fetch_trigger_names(parsed.trigger_name),
+                queries.fetch_trigger_names(pg_tg_name),
             )
             combined = "\n".join(
                 (
                     "\n".join(
                         queries.drop_trigger(t["trigger_name"], t["table"])
                         for t in trigger_names
                     ),
-                    queries.drop_function(parsed.function_name),
+                    queries.drop_function(pg_fn_name),
                 )
             )
             print(combined, flush=True)

src/pgcachewatch/listeners.py

@@ -139,7 +139,7 @@ def __init__(
     async def connect(
         self,
         connection: asyncpg.Connection,
-        channel: models.PGChannel = models.PGChannel("ch_pgcachewatch_table_change"),
+        channel: models.PGChannel = models.DEFAULT_PG_CHANNE,
     ) -> None:
         """
         Asynchronously connects the PGEventQueue to a specified
@@ -205,7 +205,7 @@ def __init__(
     async def connect(
         self,
         ws: websockets.WebSocketClientProtocol,
-        channel: models.PGChannel = models.PGChannel("ch_pgcachewatch_table_change"),
+        channel: models.PGChannel = models.DEFAULT_PG_CHANNE,
     ) -> None:
         async def _handler(ws: websockets.WebSocketClientProtocol) -> None:
             event_handler = create_event_inserter(self, self._max_latency)

src/pgcachewatch/models.py

@@ -15,6 +15,8 @@
     str,
 )
 
+DEFAULT_PG_CHANNE = PGChannel("ch_pgcachewatch_table_change")
+
 
 class DeadlineSetting(pydantic.BaseModel):
     """

src/pgcachewatch/queries.py

@@ -3,7 +3,7 @@ def create_notify_function(
     function_name: str,
 ) -> str:
     return f"""
-CREATE OR REPLACE FUNCTION {function_name}_{channel_name}() RETURNS TRIGGER AS $$
+CREATE OR REPLACE FUNCTION {function_name}() RETURNS TRIGGER AS $$
   BEGIN
     PERFORM pg_notify(
       '{channel_name}',
@@ -19,15 +19,14 @@ def create_notify_function(
 
 
 def create_after_change_trigger(
+    trigger_name: str,
     table_name: str,
-    channel_name: str,
     function_name: str,
-    trigger_name_prefix: str,
 ) -> str:
     return f"""
-CREATE OR REPLACE TRIGGER {trigger_name_prefix}{table_name}
+CREATE OR REPLACE TRIGGER {trigger_name}
   AFTER INSERT OR UPDATE OR DELETE OR TRUNCATE ON {table_name}
-  EXECUTE FUNCTION {function_name}_{channel_name}();
+  EXECUTE FUNCTION {function_name}();
 """