Changed model format and initial migration index

- Use the double-dict format for model im-/export, create_initial_data,
get_everything request.
- Use -1 as the default migration index (previously it was 1 if not the
import tools were used).

Supersedes #144

cli/create_initial_data.py

@@ -58,8 +58,8 @@ def main():
             migration_index = models
             continue
 
-        for model in models:
-            fqid = fqid_from_collection_and_id(collection, model["id"])
+        for id, model in models.items():
+            fqid = fqid_from_collection_and_id(collection, id)
             event = RequestCreateEvent(fqid, model)
             events.append(event)
 

cli/export_data_only.py

@@ -9,6 +9,7 @@
     DATASTORE_DEV_MODE_ENVIRONMENT_VAR,
     EnvironmentService,
 )
+from datastore.shared.util import is_reserved_field
 
 
 def main():
@@ -24,9 +25,9 @@ def main():
 
     # strip meta fields
     for collection, models in response.items():
-        for model in models:
+        for model in models.values():
             for field in list(model.keys()):
-                if field.startswith("meta_"):
+                if is_reserved_field(field):
                     del model[field]
 
     response["_migration_index"] = cast(Any, migration_index)

cli/import_data_only.py

@@ -35,8 +35,8 @@ def main():
             migration_index = models
             continue
 
-        for model in models:
-            fqid = fqid_from_collection_and_id(collection, model["id"])
+        for id, model in models.items():
+            fqid = fqid_from_collection_and_id(collection, id)
             event = RequestCreateEvent(fqid, model)
             events.append(event)
 

datastore/reader/core/reader.py

@@ -60,7 +60,7 @@ def get_all(self, request: GetAllRequest) -> Dict[Id, Model]:
 
     def get_everything(
         self, request: GetEverythingRequest
-    ) -> Dict[Collection, List[Model]]:
+    ) -> Dict[Collection, Dict[Id, Model]]:
         """
         Returns all models In the form of the example data: Collections mapped to
         lists of models.

datastore/reader/core/reader_service.py

@@ -132,7 +132,7 @@ def get_all(self, request: GetAllRequest) -> Dict[Id, Model]:
     @retry_on_db_failure
     def get_everything(
         self, request: GetEverythingRequest
-    ) -> Dict[Collection, List[Model]]:
+    ) -> Dict[Collection, Dict[Id, Model]]:
         return self.database.get_everything(request.get_deleted_models)
 
     @retry_on_db_failure

datastore/shared/postgresql_backend/sql_read_database_backend_service.py

@@ -112,25 +112,22 @@ def get_all(
     def get_everything(
         self,
         get_deleted_models: DeletedModelsBehaviour = DeletedModelsBehaviour.NO_DELETED,
-    ) -> Dict[Collection, List[Model]]:
+    ) -> Dict[Collection, Dict[Id, Model]]:
         del_cond = self.query_helper.get_deleted_condition(
             get_deleted_models, prepend_and=False
         )
         query = f"""
             select fqid as __fqid__, data from models
             {"where " + del_cond if del_cond else ""}"""
-
         result = self.connection.query(query, [], [])
-        unsorted_data = defaultdict(list)
+
+        data: Dict[Collection, Dict[Id, Model]] = defaultdict(dict)
         for row in result:
             collection, id = collection_and_id_from_fqid(row["__fqid__"])
             model = row["data"]
             model["id"] = id
-            unsorted_data[collection].append(model)
+            data[collection][id] = model
 
-        data = {}
-        for collection, models in unsorted_data.items():
-            data[collection] = sorted(models, key=lambda model: model["id"])
         return data
 
     def filter(
@@ -354,7 +351,7 @@ def get_current_migration_index(self) -> int:
                         + f"different to {max_migration_index}"
                     )
             else:
-                max_migration_index = 1
+                max_migration_index = -1
             self.current_migration_index = max_migration_index
         return self.current_migration_index
 

datastore/shared/services/read_database.py

@@ -83,7 +83,7 @@ def get_all(
     def get_everything(
         self,
         get_deleted_models: DeletedModelsBehaviour = DeletedModelsBehaviour.NO_DELETED,
-    ) -> Dict[Collection, List[Model]]:
+    ) -> Dict[Collection, Dict[Id, Model]]:
         """
         Returns all models of the given collection. WARNING: May result in a huge
         amount of data. Use with caution!

tests/migrations/system/conftest.py

@@ -1,3 +1,5 @@
+from unittest.mock import MagicMock
+
 import pytest
 
 from datastore.migrations import MigrationHandler, setup as migration_setup
@@ -145,3 +147,19 @@ def _assert_finalized():
         assert_count("migration_positions", 0)
 
     return _assert_finalized
+
+
+@pytest.fixture()
+def set_migration_index_to_1(migration_handler):
+    def _set_migration_index_to_1():
+        previous_logger = migration_handler.logger.info
+        migration_handler.logger.info = i = MagicMock()
+        migration_handler.migrate()  # set target migration index to 1
+        i.assert_called()
+        assert (
+            "The datastore has a migration index of -1. Set the migration index to 1."
+            in i.call_args.args[0]
+        )
+        migration_handler.logger.info = previous_logger
+
+    yield _set_migration_index_to_1

tests/migrations/system/test_basic_migrate_cases.py

@@ -12,47 +12,62 @@
 def test_no_migrations_to_apply(
     migration_handler,
     write,
+    set_migration_index_to_1,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {}})
+    set_migration_index_to_1()
 
     migration_handler.register_migrations(get_noop_migration(2))
+    migration_handler.logger.info = i = MagicMock()
     migration_handler.finalize()
 
-    migration_handler.logger.info = i = MagicMock()
+    i.assert_called()
+    assert "Position 1 from MI 1 to MI 2" in i.call_args_list[1].args[0]
+
+    i.reset_mock()
     migration_handler.migrate()
 
     i.assert_called()
     assert (
         "No migrations to apply. The productive database is up to date."
-        in i.call_args[0][0]
+        in i.call_args.args[0]
     )
 
 
 def test_finalizing_needed(
     migration_handler,
     write,
+    set_migration_index_to_1,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {}})
+    set_migration_index_to_1()
 
     migration_handler.register_migrations(get_noop_migration(2))
+    migration_handler.logger.info = i = MagicMock()
     migration_handler.migrate()
 
-    migration_handler.logger.info = i = MagicMock()
+    i.assert_called()
+    assert "Position 1 from MI 1 to MI 2" in i.call_args_list[1].args[0]
+    assert "Done. Finalizing is still be needed." in i.call_args.args[0]
+
+    i.reset_mock()
     migration_handler.migrate()
 
     i.assert_called()
     assert (
         "No migrations to apply, but finalizing is still needed."
-        in i.call_args_list[1][0][0]
+        in i.call_args_list[1].args[0]
     )
-    assert "Done. Finalizing is still be needed." in i.call_args_list[2][0][0]
+    assert "Done. Finalizing is still be needed." in i.call_args.args[0]
 
 
 def test_finalizing_not_needed(
     migration_handler,
     write,
+    set_migration_index_to_1,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {}})
+    set_migration_index_to_1()
 
     migration_handler.register_migrations(get_noop_migration(2))
     migration_handler.finalize()
@@ -73,12 +88,6 @@ def test_invalid_migration_index(
 ):
     write({"type": "create", "fqid": "a/1", "fields": {}})
 
-    with connection_handler.get_connection_context():
-        connection_handler.execute(
-            "update positions set migration_index=%s",
-            [-1],
-        )
-
     migrater = injector.get(Migrater)
 
     with pytest.raises(MismatchingMigrationIndicesException) as e:

tests/migrations/system/test_errors_during_migration.py

@@ -26,13 +26,20 @@ def fail_handler(event):
 
 
 def test_failing_migration(
-    migration_handler, write, read_model, assert_model, assert_finalized
+    migration_handler,
+    write,
+    set_migration_index_to_1,
+    read_model,
+    assert_model,
+    assert_finalized,
 ):
-    fail_migration = get_lambda_migration(do_raise(AbortException()))
-    migration_handler.register_migrations(fail_migration)
     write({"type": "create", "fqid": "a/1", "fields": {"f": 1}})
+    set_migration_index_to_1()
     previous_model = read_model("a/1")
 
+    fail_migration = get_lambda_migration(do_raise(AbortException()))
+    migration_handler.register_migrations(fail_migration)
+
     with pytest.raises(AbortException):
         migration_handler.migrate()
 
@@ -50,17 +57,20 @@ def test_failing_migration_multi_positions(
     connection_handler,
     migration_handler,
     write,
+    set_migration_index_to_1,
     read_model,
     assert_model,
     assert_finalized,
     assert_count,
 ):
-    migration_handler.register_migrations(get_lambda_migration(fail_handler))
     write({"type": "create", "fqid": "a/1", "fields": {"f": 1}})
     write({"type": "create", "fqid": "a/2", "fields": {"f": 1}})
     write({"type": "create", "fqid": "a/3", "fields": {"f": 1}})
+    set_migration_index_to_1()
     previous_models = (read_model("a/1"), read_model("a/2"), read_model("a/3"))
 
+    migration_handler.register_migrations(get_lambda_migration(fail_handler))
+
     with pytest.raises(AbortException):
         migration_handler.migrate()
 
@@ -87,18 +97,21 @@ def test_failing_migration_multi_positions_new_migration_after_fail(
     connection_handler,
     migration_handler,
     write,
+    set_migration_index_to_1,
     read_model,
     assert_model,
     assert_finalized,
     assert_count,
     query_single_value,
 ):
-    migration_handler.register_migrations(get_lambda_migration(fail_handler))
     write({"type": "create", "fqid": "a/1", "fields": {"f": 1}})
     write({"type": "create", "fqid": "a/2", "fields": {"f": 1}})
     write({"type": "create", "fqid": "a/3", "fields": {"f": 1}})
+    set_migration_index_to_1()
     previous_models = (read_model("a/1"), read_model("a/2"), read_model("a/3"))
 
+    migration_handler.register_migrations(get_lambda_migration(fail_handler))
+
     with pytest.raises(AbortException):
         migration_handler.migrate()
 
@@ -129,13 +142,15 @@ def test_use_basemigration(
     connection_handler,
     migration_handler,
     write,
+    set_migration_index_to_1,
     read_model,
     assert_model,
     assert_finalized,
     assert_count,
     query_single_value,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {"f": 1}})
+    set_migration_index_to_1()
 
     class MyBaseMigration(BaseMigration):
         target_migration_index = 2

tests/migrations/system/test_event_overwrite.py

@@ -5,8 +5,11 @@
 from ..util import get_lambda_migration, get_noop_migration
 
 
-def test_returning_original_events(migration_handler, connection_handler, write):
+def test_returning_original_events(
+    migration_handler, connection_handler, write, set_migration_index_to_1
+):
     write({"type": "create", "fqid": "a/1", "fields": {"f": 1}})
+    set_migration_index_to_1()
 
     with connection_handler.get_connection_context():
         original_events = connection_handler.query(
@@ -24,12 +27,14 @@ def test_returning_original_events(migration_handler, connection_handler, write)
 def test_new_events(
     migration_handler,
     connection_handler,
-    assert_count,
     write,
+    set_migration_index_to_1,
+    assert_count,
     assert_model,
     exists_model,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {}})
+    set_migration_index_to_1()
 
     new_events = [CreateEvent(f"a/{i}", {}) for i in (2, 3, 4)]
     migration_handler.register_migrations(get_lambda_migration(lambda e: new_events))
@@ -46,13 +51,15 @@ def test_new_events(
 def test_new_events_multiple_positions(
     migration_handler,
     connection_handler,
-    assert_count,
     write,
+    set_migration_index_to_1,
+    assert_count,
     assert_model,
     exists_model,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {}})
     write({"type": "create", "fqid": "b/1", "fields": {}})
+    set_migration_index_to_1()
 
     a_new_events = [CreateEvent(f"a/{i}", {}) for i in (2, 3)]
     b_new_events = [CreateEvent(f"b/{i}", {}) for i in (2, 3)]
@@ -76,12 +83,14 @@ def test_new_events_rebuilding_order(
     migration_handler,
     connection_handler,
     write,
+    set_migration_index_to_1,
     assert_model,
     exists_model,
     assert_count,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {"f": 1}})
     write({"type": "update", "fqid": "a/1", "fields": {"f": 2}})
+    set_migration_index_to_1()
 
     create_new_events = [
         CreateEvent("a/1", {"f": 4}),
@@ -112,11 +121,13 @@ def test_less_events(
     migration_handler,
     assert_count,
     write,
+    set_migration_index_to_1,
     assert_model,
     exists_model,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {"f": 1}})
     write({"type": "update", "fqid": "a/1", "fields": {"f": None, "f2": 2}})
+    set_migration_index_to_1()
     assert_count("events", 3)
 
     migration_handler.register_migrations(
@@ -138,10 +149,12 @@ def test_return_none_with_modifications(
     migration_handler,
     assert_count,
     write,
+    set_migration_index_to_1,
     assert_model,
     read_model,
 ):
     write({"type": "create", "fqid": "a/1", "fields": {"f": 1}})
+    set_migration_index_to_1()
     previous_model = read_model("a/1")
 
     def handler(event):