feat(signals): Add mass update method for signals (#26)

* feat(signals): Add mass_update_signals method

* test(storage): Add tests for mass_update_signals

CHANGELOG.md

@@ -10,6 +10,16 @@ functions provided by the `src/cscapi` folder.
 
 --- 
 
+## [0.5.0](https://github.com/crowdsecurity/python-capi-sdk/releases/tag/v0.5.0) - 2024-03-??
+[_Compare with previous release_](https://github.com/crowdsecurity/python-capi-sdk/compare/v0.4.0...v0.5.0)
+
+
+### Changed
+
+- **Breaking change**: Add `StorageInterface::mass_update_signals` method
+
+---
+
 ## [0.4.0](https://github.com/crowdsecurity/python-capi-sdk/releases/tag/v0.4.0) - 2024-02-23
 [_Compare with previous release_](https://github.com/crowdsecurity/python-capi-sdk/compare/v0.3.0...v0.4.0)
 

src/cscapi/client.py

@@ -24,7 +24,7 @@
 CAPI_SIGNALS_ENDPOINT = "/signals"
 CAPI_DECISIONS_ENDPOINT = "/decisions/stream"
 CAPI_METRICS_ENDPOINT = "/metrics"
-SIGNAL_BATCH_LIMIT = 1000
+SIGNAL_BATCH_LIMIT = 5000
 
 
 def has_valid_token(
@@ -237,12 +237,10 @@ def _send_signals_to_capi(
         return result
 
     def _mark_signals_as_sent(self, signals: Tuple[SignalModel]) -> List[int]:
-        result = []
-        for signal in signals:
-            self.storage.update_or_create_signal(replace(signal, sent=True))
-            result.append(signal.alert_id)
+        signal_ids = [signal.alert_id for signal in signals]
+        self.storage.mass_update_signals(signal_ids=signal_ids, changes={"sent": True})
 
-        return result
+        return signal_ids
 
     def _send_metrics_for_machine(self, machine: MachineModel):
         for _ in range(self.max_retries + 1):

src/cscapi/sql_storage.py

@@ -223,6 +223,12 @@ def update_or_create_machine(self, machine: storage.MachineModel) -> bool:
             session.execute(update_stmt)
         return False
 
+    def mass_update_signals(self, signal_ids: List[int], changes: dict):
+        with self.session.begin() as session:
+            stmt = update(SignalDBModel).where(SignalDBModel.alert_id.in_(signal_ids))
+            stmt = stmt.values(changes)
+            session.execute(stmt)
+
     def update_or_create_signal(self, signal: storage.SignalModel) -> bool:
         to_insert = SignalDBModel(
             **{

src/cscapi/storage.py

@@ -110,6 +110,11 @@ def update_or_create_signal(self, signal: SignalModel) -> bool:
         # returns true if created new row else false
         raise NotImplementedError
 
+    @abstractmethod
+    def mass_update_signals(self, signal_ids: List[int], changes: dict):
+        # changes is a dictionary with keys as the attributes to update and values as the new values
+        raise NotImplementedError
+
     @abstractmethod
     def delete_signals(self, signal_ids: List[int]):
         raise NotImplementedError

tests/test_sql_storage.py

@@ -289,3 +289,27 @@ def test_update_signal(self):
         signal = signals[0]
 
         assert signal.sent == True
+
+    def test_mass_update_signals(self):
+        assert self.storage.get_signals(limit=1000) == []
+
+        for x in range(10):
+            self.storage.update_or_create_signal(mock_signals()[0])
+
+        signals = self.storage.get_signals(limit=1000)
+
+        assert len(signals) == 10
+        for s in signals:
+            assert s.sent == False
+            assert s.scenario_trust == "trusted"
+        signal_ids = [s.alert_id for s in signals]
+        self.storage.mass_update_signals(
+            signal_ids, {"sent": True, "scenario_trust": "manual"}
+        )
+
+        signals = self.storage.get_signals(limit=1000)
+
+        assert len(signals) == 10
+        for s in signals:
+            assert s.sent == True
+            assert s.scenario_trust == "manual"