Add async materialized view refresh jobs

Adds refresh jobs for existing materialized views and invokes them:
- Refreshes the published variants materialized view whenever a new score set
is published. (We cannot delete a published score set, nor edit it's variants,
so publication time is the only moment this view becomes outdated).
- Refreshes all materialized views at 0300.

Adds tests for newly placed publication view refresh.

Clarifies connection -> inspector flow during mat view refresh.

Opens #405.

Author: bencap

src/mavedb/db/view.py

@@ -50,7 +50,7 @@ def _drop_view(element: DropView, compiler, **kw):
     return "DROP %s %s" % ("MATERIALIZED VIEW" if element.materialized else "VIEW", element.name)
 
 
-def view_exists(ddl: CreateView, target, connection: Session, materialized: bool, **kw):
+def view_exists(ddl: CreateView, target, connection: sa.Connection, materialized: bool, **kw):
     inspector = sa.inspect(connection)
     if inspector is None:
         return False
@@ -59,7 +59,7 @@ def view_exists(ddl: CreateView, target, connection: Session, materialized: bool
     return ddl.name in view_names
 
 
-def view_doesnt_exist(ddl: CreateView, target, connection: Session, materialized: bool, **kw):
+def view_doesnt_exist(ddl: CreateView, target, connection: sa.Connection, materialized: bool, **kw):
     return not view_exists(ddl, target, connection, materialized, **kw)
 
 
@@ -121,7 +121,7 @@ class MyView(Base):
     return t
 
 
-def refresh_mat_view(session, name, concurrently=True):
+def refresh_mat_view(session: Session, name: str, concurrently=True):
     """
     Refreshes a single materialized view, given by `name`.
     """
@@ -132,12 +132,15 @@ def refresh_mat_view(session, name, concurrently=True):
     session.execute(sa.text("REFRESH MATERIALIZED VIEW " + _con + name))
 
 
-# TODO: untested.
-def refresh_all_mat_views(session, concurrently=True):
+def refresh_all_mat_views(session: Session, concurrently=True):
     """
     Refreshes all materialized views. Views are refreshed in non-deterministic order,
     so view definitions can't depend on each other.
     """
-    mat_views = session.inspect(session.engine).get_view_names()
-    for v in mat_views:
-        refresh_mat_view(session, v, concurrently)
+    inspector = sa.inspect(session.connection())
+
+    if not inspector:
+        return
+
+    for mv in inspector.get_materialized_view_names():
+        refresh_mat_view(session, mv, concurrently)

src/mavedb/routers/score_sets.py

@@ -658,7 +658,6 @@ async def upload_score_set_variant_data(
     except UnicodeDecodeError as e:
         raise HTTPException(status_code=400, detail=f"Error decoding file: {e}. Ensure the file has correct values.")
 
-
     if scores_file:
         # Although this is also updated within the variant creation job, update it here
         # as well so that we can display the proper UI components (queue invocation delay
@@ -1016,11 +1015,12 @@ async def delete_score_set(
     response_model=score_set.ScoreSet,
     response_model_exclude_none=True,
 )
-def publish_score_set(
+async def publish_score_set(
     *,
     urn: str,
     db: Session = Depends(deps.get_db),
     user_data: UserData = Depends(require_current_user),
+    worker: ArqRedis = Depends(deps.get_worker),
 ) -> Any:
     """
     Publish a score set.
@@ -1097,4 +1097,18 @@ def publish_score_set(
     db.commit()
     db.refresh(item)
 
+    # await the insertion of this job into the worker queue, not the job itself.
+    job = await worker.enqueue_job(
+        "refresh_published_variants_view",
+        correlation_id_for_context(),
+        user_data.user.id,
+    )
+    if job is not None:
+        save_to_logging_context({"worker_job_id": job.job_id})
+        logger.info(msg="Enqueud published variant materialized view refresh job.", extra=logging_context())
+    else:
+        logger.warning(
+            msg="Failed to enqueue published variant materialized view refresh job.", extra=logging_context()
+        )
+
     return item

src/mavedb/worker/jobs.py

@@ -14,6 +14,7 @@
 from sqlalchemy.orm import Session
 
 from mavedb.data_providers.services import vrs_mapper
+from mavedb.db.view import refresh_all_mat_views
 from mavedb.lib.exceptions import MappingEnqueueError, NonexistentMappingReferenceError, NonexistentMappingResultsError
 from mavedb.lib.logging.context import format_raised_exception_info_as_dict
 from mavedb.lib.score_sets import (
@@ -29,6 +30,7 @@
 from mavedb.models.enums.mapping_state import MappingState
 from mavedb.models.enums.processing_state import ProcessingState
 from mavedb.models.mapped_variant import MappedVariant
+from mavedb.models.published_variant import PublishedVariantsMV
 from mavedb.models.score_set import ScoreSet
 from mavedb.models.user import User
 from mavedb.models.variant import Variant
@@ -50,7 +52,9 @@ async def mapping_in_execution(redis: ArqRedis, job_id: str):
         await redis.set(MAPPING_CURRENT_ID_NAME, "")
 
 
-def setup_job_state(ctx, invoker: int, resource: Optional[str], correlation_id: str):
+def setup_job_state(
+    ctx, invoker: Optional[int], resource: Optional[str], correlation_id: Optional[str]
+) -> dict[str, Any]:
     ctx["state"][ctx["job_id"]] = {
         "application": "mavedb-worker",
         "user": invoker,
@@ -653,3 +657,20 @@ async def variant_mapper_manager(ctx: dict, correlation_id: str, updater_id: int
         db.commit()
 
         return {"success": False, "enqueued_job": new_job_id}
+
+
+# TODO#405: Refresh materialized views within an executor.
+async def refresh_materialized_views(ctx: dict):
+    logging_context = setup_job_state(ctx, None, None, None)
+    logger.debug(msg="Began refresh materialized views.", extra=logging_context)
+    refresh_all_mat_views(ctx["db"])
+    logger.debug(msg="Done refreshing materialized views.", extra=logging_context)
+    return {"success": True}
+
+
+async def refresh_published_variants_view(ctx: dict, correlation_id: str):
+    logging_context = setup_job_state(ctx, None, None, correlation_id)
+    logger.debug(msg="Began refresh of published variants materialized view.", extra=logging_context)
+    PublishedVariantsMV.refresh(ctx["db"])
+    logger.debug(msg="Done refreshing of published variants materialized view.", extra=logging_context)
+    return {"success": True}

src/mavedb/worker/settings.py

@@ -3,20 +3,29 @@
 from typing import Callable
 
 from arq.connections import RedisSettings
-from arq.cron import CronJob
+from arq.cron import CronJob, cron
 
 from mavedb.data_providers.services import cdot_rest
 from mavedb.db.session import SessionLocal
 from mavedb.lib.logging.canonical import log_job
-from mavedb.worker.jobs import create_variants_for_score_set, map_variants_for_score_set, variant_mapper_manager
+from mavedb.worker.jobs import (
+    create_variants_for_score_set,
+    map_variants_for_score_set,
+    variant_mapper_manager,
+    refresh_materialized_views,
+    refresh_published_variants_view,
+)
 
 # ARQ requires at least one task on startup.
 BACKGROUND_FUNCTIONS: list[Callable] = [
     create_variants_for_score_set,
     variant_mapper_manager,
     map_variants_for_score_set,
+    refresh_published_variants_view,
+]
+BACKGROUND_CRONJOBS: list[CronJob] = [
+    cron(refresh_materialized_views, name="refresh_all_materialized_views", hour=3, minute=0)
 ]
-BACKGROUND_CRONJOBS: list[CronJob] = []
 
 REDIS_IP = os.getenv("REDIS_IP") or "localhost"
 REDIS_PORT = int(os.getenv("REDIS_PORT") or 6379)

tests/helpers/util.py

@@ -240,8 +240,10 @@ def create_acc_score_set_with_variants(
 
 
 def publish_score_set(client, score_set_urn):
-    response = client.post(f"/api/v1/score-sets/{score_set_urn}/publish")
-    assert response.status_code == 200, f"Could not publish score set {score_set_urn}"
+    with patch.object(ArqRedis, "enqueue_job", return_value=None) as worker_queue:
+        response = client.post(f"/api/v1/score-sets/{score_set_urn}/publish")
+        assert response.status_code == 200, f"Could not publish score set {score_set_urn}"
+        worker_queue.assert_called_once()
 
     response_data = response.json()
     jsonschema.validate(instance=response_data, schema=ScoreSet.schema())