Add util for refreshing specific mat views

Author: mikicz

django_pgviews/refresh.py

@@ -0,0 +1,42 @@
+__all__ = ["refresh_specific_views"]
+
+import logging
+from collections.abc import Iterable
+
+from django_pgviews import view as pg
+from django_pgviews.dependencies import get_views_dependencies, get_views_dependendants, reorder_by_dependencies
+
+logger = logging.getLogger("django_pgviews.refresh")
+
+
+def refresh_specific_views(
+    to_refresh: Iterable[type[pg.View]],
+    concurrently: bool = True,
+    strict: bool = False,
+) -> int:
+    """
+    For a specific set of views, refresh all the materialized views, including all dependants and dependencies.
+
+    Returns the number of materialized views refreshed.
+    """
+
+    to_refresh = list(to_refresh)
+
+    to_refresh.extend(get_views_dependendants(to_refresh))
+    to_refresh.extend(get_views_dependencies(to_refresh))
+    to_refresh = reorder_by_dependencies(to_refresh)
+
+    to_refresh_mat_views: list[type[pg.MaterializedView]] = [
+        x for x in to_refresh if issubclass(x, pg.MaterializedView)
+    ]
+
+    count = 0
+
+    for model in to_refresh_mat_views:
+        model.refresh(concurrently=concurrently, strict=strict)
+
+        logger.info(f"Refreshed {model._meta.label}")
+
+        count += 1
+
+    return count

tests/test_project/viewtest/test_views.py

@@ -20,7 +20,6 @@
 from django_pgviews.signals import all_views_synced, view_synced
 
 from . import models
-from .models import LatestSuperusers
 
 try:
     from psycopg.errors import UndefinedTable
@@ -252,7 +251,7 @@ def test_get_sql(self):
 
         call_command("sync_pgviews", update=False)
 
-        assert LatestSuperusers.objects.count() == 1
+        assert models.LatestSuperusers.objects.count() == 1
 
     def test_sync_pgviews_materialized_views_check_sql_changed_disabled(self, settings: SettingsWrapper):
         settings.MATERIALIZED_VIEWS_CHECK_SQL_CHANGED = False

tests/test_refresh.py

@@ -0,0 +1,43 @@
+import pytest
+
+from django_pgviews.exceptions import ConcurrentIndexNotDefinedError
+from django_pgviews.refresh import refresh_specific_views
+from tests.test_project.viewtest import models
+
+
+def test_refresh_no_views():
+    assert refresh_specific_views([]) == 0
+
+
+@pytest.mark.django_db
+def test_refresh():
+    assert models.MaterializedRelatedView.objects.count() == 0, "MaterializedRelatedView should not have anything"
+    assert models.DependantMaterializedView.objects.count() == 0, "DependantMaterializedView should not have anything"
+
+    test_model = models.TestModel()
+    test_model.name = "Bob"
+    test_model.save()
+
+    assert refresh_specific_views([models.MaterializedRelatedView, models.RelatedView]) == 2
+
+    assert models.MaterializedRelatedView.objects.count() == 1
+    assert models.DependantMaterializedView.objects.count() == 1
+
+
+@pytest.mark.django_db
+def test_refresh_concurrently():
+    with pytest.raises(ConcurrentIndexNotDefinedError):
+        assert refresh_specific_views(
+            [models.MaterializedRelatedView, models.RelatedView, models.MaterializedRelatedViewWithIndex],
+            concurrently=True,
+            strict=True,
+        )
+
+    assert (
+        refresh_specific_views(
+            [models.MaterializedRelatedView, models.RelatedView, models.MaterializedRelatedViewWithIndex],
+            concurrently=True,
+            strict=False,
+        )
+        == 3
+    )