_run_once returns bool, run() translates to exit code

Author: koenvo

ingestify/infra/event_log/consumer.py

@@ -77,14 +77,13 @@ def _update_cursor(self, conn, event_id: int) -> None:
         )
         conn.commit()
 
-    def _run_once(self, on_event: Callable, batch_size: int = 100) -> int:
+    def _run_once(self, on_event: Callable, batch_size: int = 100) -> bool:
+        """Returns True on success, False if a processing error occurred."""
         with self._engine.connect() as conn:
             self._ensure_reader_state(conn)
             last_id = self._get_last_event_id(conn)
 
         rows = self._event_log.fetch_batch(last_id, batch_size)
-        if not rows:
-            return 0
 
         with self._engine.connect() as conn:
             for event_id, event in rows:
@@ -96,10 +95,10 @@ def _run_once(self, on_event: Callable, batch_size: int = 100) -> int:
                         event_id,
                         type(event).event_type,
                     )
-                    return 1
+                    return False
                 self._update_cursor(conn, event_id)
 
-        return 0
+        return True
 
     def run(
         self,
@@ -108,7 +107,7 @@ def run(
         batch_size: int = 100,
     ) -> int:
         while True:
-            exit_code = self._run_once(on_event, batch_size)
-            if exit_code != 0 or poll_interval is None:
-                return exit_code
+            success = self._run_once(on_event, batch_size)
+            if not success or poll_interval is None:
+                return 0 if success else 1
             time.sleep(poll_interval)

ingestify/tests/test_event_log.py

@@ -80,8 +80,7 @@ def test_consumer_processes_events(consumer, dataset):
 
 def test_consumer_cursor_not_advanced_on_error(consumer, dataset):
     consumer._event_log.write(RevisionAdded(dataset=dataset))
-    exit_code = consumer._run_once(lambda e: 1 / 0)
-    assert exit_code == 1
+    assert not consumer._run_once(lambda e: 1 / 0)
     # Next run still sees the same event
     received = []
     consumer._run_once(lambda e: received.append(e))