Add a check to the loader to make sure that the loader is compatible with the target database (#125)

Author: drnextgis

pypgstac/pypgstac/load.py

@@ -37,6 +37,7 @@
 
 from .db import PgstacDB
 from .hydration import dehydrate
+from .version import __version__
 from enum import Enum
 
 logger = logging.getLogger(__name__)
@@ -159,6 +160,13 @@ def __init__(self, db: PgstacDB):
         self.db = db
         self._partition_cache: Dict[str, Partition] = {}
 
+    def check_version(self) -> None:
+        if self.db.version != __version__:
+            raise Exception(
+                f"pypgstac version {__version__} is not compatible with the target"
+                f" database version {self.db.version}."
+            )
+
     @lru_cache
     def collection_json(self, collection_id: str) -> Tuple[Dict[str, Any], int, str]:
         """Get collection."""
@@ -183,6 +191,8 @@ def load_collections(
         insert_mode: Optional[Methods] = Methods.insert,
     ) -> None:
         """Load a collections json or ndjson file."""
+        self.check_version()
+
         if file is None:
             file = "stdin"
         conn = self.db.connect()
@@ -555,6 +565,8 @@ def load_items(
         chunksize: Optional[int] = 10000,
     ) -> None:
         """Load items json records."""
+        self.check_version()
+
         if file is None:
             file = "stdin"
         t = time.perf_counter()

pypgstac/tests/test_load.py

@@ -1,6 +1,7 @@
 """Tests for pypgstac."""
 import json
 from pathlib import Path
+from unittest import mock
 from pypgstac.load import Methods, Loader, read_json
 from psycopg.errors import UniqueViolation
 import pytest
@@ -339,3 +340,29 @@ def test_load_dehydrated(loader: Loader) -> None:
     loader.load_items(
         str(dehydrated_items), insert_mode=Methods.insert, dehydrated=True
     )
+
+
+def test_load_collections_incompatible_version(loader: Loader) -> None:
+    """Test pypgstac collections loader raises an exception for incompatible version."""
+    with mock.patch(
+        "pypgstac.db.PgstacDB.version", new_callable=mock.PropertyMock
+    ) as mock_version:
+        mock_version.return_value = "dummy"
+        with pytest.raises(Exception):
+            loader.load_collections(
+                str(TEST_COLLECTIONS_JSON),
+                insert_mode=Methods.insert,
+            )
+
+
+def test_load_items_incompatible_version(loader: Loader) -> None:
+    """Test pypgstac items loader raises an exception for incompatible version."""
+    with mock.patch(
+        "pypgstac.db.PgstacDB.version", new_callable=mock.PropertyMock
+    ) as mock_version:
+        mock_version.return_value = "dummy"
+        with pytest.raises(Exception):
+            loader.load_items(
+                str(TEST_ITEMS),
+                insert_mode=Methods.insert,
+            )