migrator: convert "op" to OperationType enum

migrator/tests/extract/test_kafka_extract.py

@@ -15,6 +15,7 @@
 from unittest.mock import PropertyMock
 
 from invenio_rdm_migrator.extract import Tx
+from invenio_rdm_migrator.load.postgresql.transactions.operations import OperationType
 
 from zenodo_rdm_migrator.extract import KafkaExtract, KafkaExtractEnd
 
@@ -104,6 +105,9 @@ def _assert_result(
 ):
     assert len(result) == count
     assert all(isinstance(t, Tx) for t in result)
+    assert all(
+        all(isinstance(o["op"], OperationType) for o in t.operations) for t in result
+    )
     tx_dict = {t.id: t for t in result}
     tx_ids = list(tx_dict.keys())
     assert tx_ids == sorted(tx_ids)

migrator/zenodo_rdm_migrator/extract/kafka.py

@@ -15,6 +15,7 @@
 
 import dictdiffer
 from invenio_rdm_migrator.extract import Extract, Tx
+from invenio_rdm_migrator.load.postgresql.transactions.operations import OperationType
 from invenio_rdm_migrator.logging import Logger
 from kafka import KafkaConsumer, TopicPartition
 from sortedcontainers import SortedDict, SortedList
@@ -53,6 +54,8 @@ def info(self, val):
 
     def append(self, op):
         """Add a single table row operation to the transaction state."""
+        # Convert the "op" key to an enum
+        op["op"] = OperationType(op["op"].upper())
         self.ops.add(op)
 
         # Update table row counts with the operations so far