Merge pull request #577 from DataIntegrationGroup/fix/pg-cron-optional

pg_cron should be optional for local development

Author: jirhiker

alembic/versions/d5e6f7a8b9c0_create_pygeoapi_supporting_views.py

@@ -314,12 +314,8 @@ def upgrade() -> None:
             ")"
         )
     ).scalar()
-    if not pg_cron_available:
-        raise RuntimeError(
-            "Cannot schedule nightly pygeoapi materialized view refresh job: "
-            "pg_cron extension is not available on this PostgreSQL server."
-        )
-    op.execute(text("CREATE EXTENSION IF NOT EXISTS pg_cron"))
+    if pg_cron_available:
+        op.execute(text("CREATE EXTENSION IF NOT EXISTS pg_cron"))
 
     for view_id, thing_type in THING_COLLECTIONS:
         safe_view_id = _safe_view_id(view_id)
@@ -364,11 +360,21 @@ def upgrade() -> None:
     _create_matview_indexes()
 
     op.execute(text(_create_refresh_function()))
-    op.execute(text(_schedule_refresh_job()))
+    if pg_cron_available:
+        op.execute(text(_schedule_refresh_job()))
 
 
 def downgrade() -> None:
-    op.execute(text(_unschedule_refresh_job()))
+    bind = op.get_bind()
+    pg_cron_available = bind.execute(
+        text(
+            "SELECT EXISTS ("
+            "SELECT 1 FROM pg_available_extensions WHERE name = 'pg_cron'"
+            ")"
+        )
+    ).scalar()
+    if pg_cron_available:
+        op.execute(text(_unschedule_refresh_job()))
     op.execute(text(f"DROP FUNCTION IF EXISTS public.{REFRESH_FUNCTION_NAME}()"))
     _drop_view_or_materialized_view("ogc_avg_tds_wells")
     _drop_view_or_materialized_view("ogc_latest_depth_to_water_wells")

core/initializers.py

@@ -73,12 +73,8 @@ def erase_and_rebuild_db():
                 ")"
             )
         ).scalar()
-        if not pg_cron_available:
-            raise RuntimeError(
-                "Cannot erase and rebuild database: pg_cron extension is not "
-                "available on this PostgreSQL server."
-            )
-        session.execute(text("CREATE EXTENSION IF NOT EXISTS pg_cron"))
+        if pg_cron_available:
+            session.execute(text("CREATE EXTENSION IF NOT EXISTS pg_cron"))
         session.commit()
         Base.metadata.drop_all(session.bind)
         Base.metadata.create_all(session.bind)

db/initialization.py

@@ -2,7 +2,7 @@
 
 import os
 
-from sqlalchemy import text
+from sqlalchemy import inspect as sa_inspect, text
 from sqlalchemy.engine import Connection
 from sqlalchemy.orm import Session
 from sqlalchemy_searchable import sync_trigger
@@ -69,12 +69,8 @@ def recreate_public_schema(session: Session) -> None:
             ")"
         )
     ).scalar()
-    if not pg_cron_available:
-        raise RuntimeError(
-            "Cannot initialize database schema: pg_cron extension is not available "
-            "on this PostgreSQL server."
-        )
-    session.execute(text("CREATE EXTENSION IF NOT EXISTS pg_cron"))
+    if pg_cron_available:
+        session.execute(text("CREATE EXTENSION IF NOT EXISTS pg_cron"))
     session.execute(APP_READ_GRANT_SQL)
     grant_app_read_members(session)
     session.commit()
@@ -83,7 +79,11 @@ def recreate_public_schema(session: Session) -> None:
 def sync_search_vector_triggers(session: Session) -> None:
     """Ensure SQLAlchemy-searchable triggers exist for every TSVector column."""
     conn = session.connection()
+    inspector = sa_inspect(conn)
+    existing_tables = set(inspector.get_table_names())
     for table in Base.metadata.tables.values():
+        if table.name not in existing_tables:
+            continue
         for column in table.columns:
             if isinstance(column.type, TSVectorType):
                 sync_trigger(conn, table.name, column.name, list(column.type.columns))