Add Hx tracking to climo tables, adjust functions as needed to support join tables (combined keys)

Author: Nospamas

pycds/__init__.py

@@ -70,10 +70,14 @@
     "ObsCountPerDayHistory",
     "ObsWithFlags",
     "ClimatologicalPeriod",
+    "ClimatologicalPeriodHistory",
     "ClimatologicalStation",
+    "ClimatologicalStationHistory",
     "ClimatologicalStationXHistory",
     "ClimatologicalVariable",
+    "ClimatologicalVariableHistory",
     "ClimatologicalValue",
+    "ClimatologicalValueHistory",
 ]
 
 from pycds.context import get_schema_name, get_su_role_name
@@ -100,10 +104,14 @@
     PCICFlag,
     DerivedValue,
     ClimatologicalPeriod,
+    ClimatologicalPeriodHistory,
     ClimatologicalStation,
+    ClimatologicalStationHistory,
     ClimatologicalStationXHistory,
     ClimatologicalVariable,
+    ClimatologicalVariableHistory,
     ClimatologicalValue,
+    ClimatologicalValueHistory,
 )
 
 from .orm.views import (

pycds/alembic/change_history_utils.py

@@ -63,7 +63,7 @@ def drop_history_cols_from_primary(
     op.execute(f"ALTER TABLE {main_table_name(collection_name)} {drop_columns}")
 
 
-def create_history_table(collection_name: str, foreign_tables: list[tuple[str, str]]):
+def create_history_table(collection_name: str, foreign_tables: list[tuple[str, str]] | None):
     # Create the history table. We can't use Alembic create_table here because it doesn't
     # support the LIKE syntax we need.
     columns = ", ".join(
@@ -89,18 +89,23 @@ def drop_history_table(collection_name: str):
 
 def create_history_table_indexes(
     collection_name: str,
-    pri_id_name: str,
-    foreign_tables: list[tuple[str, str]],
+    pri_id_name: list[str] | str,
+    foreign_tables: Iterable[tuple[str, str]] | None,
     extras=None,
 ):
     """
     Create indexes on the history table. For analysis on what indexes are needed,
     see https://github.com/pacificclimate/pycds/issues/228
     """
 
+    if isinstance(pri_id_name, str):
+        pri_id_name = [pri_id_name]
+
+    seen = []
+
     for columns in (
         # Index on primary table primary key, mod_time, mod_user
-        ([pri_id_name], ["mod_time"], ["mod_user"])
+        tuple([x] for x in pri_id_name) + (["mod_time"], ["mod_user"])
         # Index on all foreign main table primary keys
         + tuple([ft_pk_name] for _, ft_pk_name in (foreign_tables or tuple()))
         # Index on all foreign history table primary keys
@@ -110,6 +115,9 @@ def create_history_table_indexes(
         )
         + (extras or tuple())
     ):
+        if columns in seen:
+            continue
+        seen.append(columns)
         # How much do we care about index naming? SQLAlchemy uses a different pattern than
         # appears typical in CRMP.
         op.create_index(
@@ -122,8 +130,8 @@ def create_history_table_indexes(
 
 def populate_history_table(
     collection_name: str,
-    pri_id_name: str,
-    foreign_tables: list[tuple[str, str]],
+    pri_id_name: list[str] | str,
+    foreign_tables: list[tuple[str, str]] | None,
     limit: int | None = None,
 ):
     """
@@ -140,7 +148,7 @@ def populate_history_table(
     # foreign table definitions: the CTE names, the CTE definitions, and their usages
     # within the query that populates the target history table.
 
-    foreign_tables = foreign_tables or tuple()
+    foreign_tables = foreign_tables or []
 
     conditional_comma = "," if len(foreign_tables) > 0 else ""
 
@@ -173,6 +181,11 @@ def populate_history_table(
         else ""
     )
 
+    if isinstance(pri_id_name, str):
+        pri_id_name = [pri_id_name]
+    
+    pri_order_clause = ", ".join(f"main.{idn}" for idn in pri_id_name) 
+
     stmt = f"""
         {"WITH" if len(foreign_tables) > 0 else ""}
         {ft_cte_list}
@@ -183,7 +196,7 @@ def populate_history_table(
         FROM {main_table_name(collection_name)} main  
             {conditional_comma} {ft_cte_name_list}
         {ft_where_clause}
-        ORDER BY main.{pri_id_name}        
+        ORDER BY {pri_order_clause}
     """
     op.execute(stmt)
 
@@ -209,7 +222,7 @@ def create_primary_table_triggers(collection_name: str, prefix: str = "t100_"):
 
 
 def create_history_table_triggers(
-    collection_name: str, foreign_tables: list, prefix: str = "t100_"
+    collection_name: str, foreign_tables: list | None, prefix: str = "t100_"
 ):
     # Trigger: Add foreign key values to each record inserted into history table.
     ft_args = (

pycds/alembic/versions/7244176be9fa_apply_hx_tracking_to_multi_climo_normals.py

@@ -0,0 +1,72 @@
+"""apply hx tracking to multi climo normals
+
+Revision ID: 7244176be9fa
+Revises: 758be4f4ce0f
+Create Date: 2025-09-23 16:15:58.236278
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+from pycds.alembic.util import grant_standard_table_privileges
+from pycds.context import get_schema_name
+from pycds.alembic.change_history_utils import (
+    add_history_cols_to_primary,
+    create_history_table,
+    populate_history_table,
+    drop_history_triggers,
+    drop_history_table,
+    drop_history_cols_from_primary,
+    create_history_table_triggers,
+    create_primary_table_triggers,
+    create_history_table_indexes,
+)
+
+
+# revision identifiers, used by Alembic.
+revision = "7244176be9fa"
+down_revision = "758be4f4ce0f"
+branch_labels = None
+depends_on = None
+
+
+schema_name = get_schema_name()
+
+
+table_info = (
+    # table_name, primary_key_name, foreign_keys, extra_indexes
+    ("climo_period", "climo_period_id", None, None),
+    ("climo_station", "climo_station_id", [("climo_period", "climo_period_id"), ], None),
+    ("climo_stn_x_hist", ["climo_station_id", "history_id"], [("climo_station", "climo_station_id"), ("meta_history", "history_id")], None),
+    ("climo_variable", "climo_variable_id", None, None),
+    ("climo_value", "climo_value_id", [("climo_variable", "climo_variable_id"), ("climo_station", "climo_station_id")], None),
+)
+
+def upgrade():
+    
+    # We have to set the search_path so that the trigger functions fired when
+    # the history table is populated can find the functions that they call.
+    op.get_bind().execute(sa.text(f"SET search_path TO {schema_name}, public"))
+
+    for table_name, primary_key_name, foreign_tables, extra_indexes in table_info:
+        # Primary table
+        add_history_cols_to_primary(table_name)
+        create_primary_table_triggers(table_name)
+
+        # History table
+        create_history_table(table_name, foreign_tables)
+        populate_history_table(table_name, primary_key_name, foreign_tables)
+        # History table triggers must be created after the table is populated.
+        create_history_table_triggers(table_name, foreign_tables)
+        create_history_table_indexes(
+            table_name, primary_key_name, foreign_tables, extra_indexes
+        )
+        grant_standard_table_privileges(table_name, schema=schema_name)
+
+
+def downgrade():
+    for table_name, _, _, _ in reversed(table_info):
+        drop_history_triggers(table_name)
+        drop_history_table(table_name)
+        drop_history_cols_from_primary(table_name)

pycds/alembic/versions/758be4f4ce0f_support_multiple_climatological_normals.py

@@ -1,4 +1,4 @@
-"""Support multiple climatological normals
+"""Support multiple climo normals
 
 Revision ID: 758be4f4ce0f
 Revises: 7ab87f8fbcf4
@@ -42,16 +42,14 @@
 def upgrade():
 
     op.create_table(
-        "climatological_period",
-        Column("climatological_period_id", Integer, primary_key=True),
+        "climo_period",
+        Column("climo_period_id", Integer, primary_key=True),
         Column("start_date", DateTime, nullable=False),
         Column("end_date", DateTime, nullable=False),
-        Column("mod_time", DateTime, nullable=False),
-        Column("mod_user", String, nullable=False),
         schema=schema_name,
-    )
+    ) 
 
-    Enum("long-record", "composite", "prism", name="climatological_station_type_enum").create(op.get_bind())
+    Enum("long-record", "composite", "prism", name="climo_station_type_enum").create(op.get_bind())
 
     op.create_table(
         # TODO: Columns in this table parallel those in meta_station and meta_history.
@@ -62,35 +60,33 @@ def upgrade():
         #   prevent entry of erroneous values that just happen to be very long?)
         #
         # - None are nullable. In contrast, most in the model tables are.
-        "climatological_station",  # TODO: Revise name?
-        Column("climatological_station_id", Integer, primary_key=True),
+        "climo_station",  # TODO: Revise name?
+        Column("climo_station_id", Integer, primary_key=True),
         Column(
-            "type", PG_ENUM("long-record", "composite", "prism", name="climatological_station_type_enum", create_type=False), nullable=False
+            "type", PG_ENUM("long-record", "composite", "prism", name="climo_station_type_enum", create_type=False), nullable=False
         ),
         Column("basin_id", Integer, nullable=True),
         Column("comments", String, nullable=False),
         Column(
-            "climatological_period_id",
+            "climo_period_id",
             Integer,
             ForeignKey(
-                f"{schema_name}.climatological_period.climatological_period_id"
+                f"{schema_name}.climo_period.climo_period_id"
             ),
             nullable=False,
         ),
-        Column("mod_time", DateTime, nullable=False),
-        Column("mod_user", String, nullable=False),
         schema=schema_name,
     )
 
-    Enum("base", "joint", name="climatological_station_role_enum").create(op.get_bind())
+    Enum("base", "joint", name="climo_station_role_enum").create(op.get_bind())
 
     op.create_table(
-        "climatological_station_x_meta_history",
+        "climo_stn_x_hist",
         Column(
-            "climatological_station_id",
+            "climo_station_id",
             Integer,
             ForeignKey(
-                f"{schema_name}.climatological_station.climatological_station_id"
+                f"{schema_name}.climo_station.climo_station_id"
             ),
             primary_key=True,
         ),
@@ -100,13 +96,11 @@ def upgrade():
             ForeignKey(f"{schema_name}.meta_history.history_id"),
             primary_key=True,
         ),
-        Column("role", PG_ENUM("base", "joint", name="climatological_station_role_enum", create_type=False), nullable=False),
-        Column("mod_time", DateTime, nullable=False),
-        Column("mod_user", String, nullable=False),
+        Column("role", PG_ENUM("base", "joint", name="climo_station_role_enum", create_type=False), nullable=False),
         schema=schema_name,
     )
 
-    Enum("annual", "seasonal", "monthly", name="climatology_duration_enum").create(op.get_bind())
+    Enum("annual", "seasonal", "monthly", name="climo_duration_enum").create(op.get_bind())
 
     op.create_table(
         # TODO: Columns in this table parallel those in meta_vars.
@@ -117,54 +111,50 @@ def upgrade():
         #   prevent entry of erroneous values that just happen to be very long?)
         #
         # - None are nullable. In contrast, most in the model tables are.
-        "climatological_variable",
-        Column("climatological_variable_id", Integer, primary_key=True),
+        "climo_variable",
+        Column("climo_variable_id", Integer, primary_key=True),
         Column(
-            "duration", PG_ENUM("annual", "seasonal", "monthly", name="climatology_duration_enum", create_type=False), nullable=False
+            "duration", PG_ENUM("annual", "seasonal", "monthly", name="climo_duration_enum", create_type=False), nullable=False
         ),
         Column("unit", String, nullable=False),
         Column("standard_name", String, nullable=False),
         Column("display_name", String, nullable=False),
         Column("short_name", String, nullable=False),
         Column("cell_methods", String, nullable=False),
         Column("net_var_name", CITEXT(), nullable=False),
-        Column("mod_time", DateTime, nullable=False),
-        Column("mod_user", String, nullable=False),
         schema=schema_name,
     )
 
     op.create_table(
-        "climatological_value",
-        Column("climatological_value_id", Integer, primary_key=True),
+        "climo_value",
+        Column("climo_value_id", Integer, primary_key=True),
         Column("value_time", DateTime, nullable=False),
         Column("value", Float, nullable=False),
         Column("num_contributing_years", Integer, nullable=False),
         Column(
-            "climatological_variable_id",
+            "climo_variable_id",
             Integer,
             ForeignKey(
-                f"{schema_name}.climatological_variable.climatological_variable_id"
+                f"{schema_name}.climo_variable.climo_variable_id"
             ),
         ),
         Column(
-            "climatological_station_id",
+            "climo_station_id",
             Integer,
             ForeignKey(
-                f"{schema_name}.climatological_station.climatological_station_id"
+                f"{schema_name}.climo_station.climo_station_id"
             ),
         ),
-        Column("mod_time", DateTime, nullable=False),
-        Column("mod_user", String, nullable=False),
         schema=schema_name,
     )
 
 
 def downgrade():
-    op.drop_table("climatological_value", schema=schema_name)
-    op.drop_table("climatological_variable", schema=schema_name)
-    Enum("annual", "seasonal", "monthly", name="climatology_duration_enum").drop(op.get_bind())
-    op.drop_table("climatological_station_x_meta_history", schema=schema_name)
-    Enum("base", "joint", name="climatological_station_role_enum").drop(op.get_bind())
-    op.drop_table("climatological_station", schema=schema_name)
-    Enum("long-record", "composite", "prism", name="climatological_station_type_enum").drop(op.get_bind())
-    op.drop_table("climatological_period", schema=schema_name)
+    op.drop_table("climo_value", schema=schema_name)
+    op.drop_table("climo_variable", schema=schema_name)
+    Enum("annual", "seasonal", "monthly", name="climo_duration_enum").drop(op.get_bind())
+    op.drop_table("climo_stn_x_hist", schema=schema_name)
+    Enum("base", "joint", name="climo_station_role_enum").drop(op.get_bind())
+    op.drop_table("climo_station", schema=schema_name)
+    Enum("long-record", "composite", "prism", name="climo_station_type_enum").drop(op.get_bind())
+    op.drop_table("climo_period", schema=schema_name)