Repository: AgnostiqHQ/covalent
Pull request #1672 (2023-05-26 10:08:41)

Porting SQLite Trigger to develop
<!--
⚠️ Your pull request title should be short detailed and understandable for all.
⚠️ Increment the version number in the /VERSION file. If this is a bugfix, increment the patch version (the rightmost number), for example 0.18.2 becomes 0.18.3. If it's a new feature, increment the minor version (the middle number), for example 0.18.2 becomes 0.19.0.
⚠️ Add a note to /CHANGELOG.md with your version number and the date, summarizing the changes.
⚠️ If your pull request fixes an open issue, please link to the issue.
⚠️ Rebase the latest changes from the develop branch. Assuming origin points to https://github.com/AgnostiqHQ/covalent, you should run git rebase origin/develop.
-->

- [x] I have added the tests to cover my changes.
- [x] I have updated the documentation and CHANGELOG accordingly.
- [x] I have read the CONTRIBUTING document.

Adding the SQLite Trigger

github-actions[bot]: Hello. You may have forgotten to update the changelog!
Please edit [`CHANGELOG.md`](/AgnostiqHQ/covalent/blob/master/CHANGELOG.md) with a one-to-two sentence description of the change in the UNRELEASED section. You may include a small working example for new features.
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1672?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) Report
> Merging [#1672](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1672?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (434dc87) into [develop](https://app.codecov.io/gh/AgnostiqHQ/covalent/commit/1653d5d78485ff4d59a2aa43e49e290441e908f9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ) (1653d5d) will **increase** coverage by `1.27%`.
> The diff coverage is `49.29%`.

> :exclamation: Current head 434dc87 differs from pull request most recent head b4e5ba1. Consider uploading reports for the commit b4e5ba1 to get more accurate results

<details><summary>Additional details and impacted files</summary>


[![Impacted file tree graph](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1672/graphs/tree.svg?width=650&height=150&src=pr&token=YGHCB3DE4P&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)](https://app.codecov.io/gh/AgnostiqHQ/covalent/pull/1672?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=AgnostiqHQ)

```diff
@@             Coverage Diff             @@
##           develop    #1672      +/-   ##
===========================================
+ Coverage    87.67%   88.94%   +1.27%     
===========================================
  Files          158       83      -75     
  Lines         6740     5084    -1656     
  Branches       151        0     -151     
===========================================
- Hits          5909     4522    -1387     
+ Misses         726      562     -164     
+ Partials       105        0     -105     
```

| Flag | Coverage Δ | |
|---|---|---|
| Dispatcher | `87.51% <100.00%> (-0.06%)` | :arrow_down: |
| Functional_Tests | `?` | |
| SDK | `89.79% <48.57%> (-0.89%)` | :arrow_down: |
| UI_Frontend | `?` | |

</details>

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2ccb00d2..1e48d5fd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,6 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added ability to hide post-processing electrons on the UI.
 - Added prettify of names for the graph screen on the UI.
 - Ability to specify a `workdir` for `local` and `dask` executors along with `create_unique_workdir` option for each electron / node.
+- Added `SQLiteTrigger` class to the `triggers` module.
 
 ### Removed
 
diff --git a/covalent/triggers/__init__.py b/covalent/triggers/__init__.py
index a3875166..ca9cd77c 100644
--- a/covalent/triggers/__init__.py
+++ b/covalent/triggers/__init__.py
@@ -20,5 +20,6 @@
 
 from .base import BaseTrigger  # nopycln: import
 from .dir_trigger import DirTrigger  # nopycln: import
+from .sqlite_trigger import SQLiteTrigger  # nopycln: import
 from .time_trigger import TimeTrigger  # nopycln: import
 from .trigger_loader import available_triggers  # nopycln: import
diff --git a/covalent/triggers/dir_trigger.py b/covalent/triggers/dir_trigger.py
index ef22139a..8be7d02f 100644
--- a/covalent/triggers/dir_trigger.py
+++ b/covalent/triggers/dir_trigger.py
@@ -71,10 +71,10 @@ class DirTrigger(BaseTrigger):
         dir_path,
         event_names,
         batch_size: int = 1,
+        recursive: bool = False,
         lattice_dispatch_id: str = None,
         dispatcher_addr: str = None,
         triggers_server_addr: str = None,
-        recursive: bool = False,
     ):
         super().__init__(lattice_dispatch_id, dispatcher_addr, triggers_server_addr)
 
diff --git a/covalent/triggers/sqlite_trigger.py b/covalent/triggers/sqlite_trigger.py
new file mode 100644
index 00000000..f233d934
--- /dev/null
+++ b/covalent/triggers/sqlite_trigger.py
@@ -0,0 +1,135 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+import sqlite3
+import time
+from functools import partial
+from threading import Event
+from typing import List
+
+from covalent._shared_files import logger
+
+from .base import BaseTrigger
+
+app_log = logger.app_log
+log_stack_info = logger.log_stack_info
+
+
+class SQLiteTrigger(BaseTrigger):
+    """
+    SQLite based Trigger which can read for changes in a SQLite database
+    and trigger workflows based on that.
+
+    Args:
+        db_path: Absolute path to the database file
+        table_name: Name of the table to observe
+        poll_interval: Time in seconds to wait for before reading the database again
+        where_clauses: List of "WHERE" conditions, e.g. ["id > 2", "status = pending"], to check when
+                       polling the database
+        trigger_after_n: Number of times the event must happen after which the workflow will be triggered.
+                         e.g value of 2 means workflow will be triggered once the event has occurred twice.
+        lattice_dispatch_id: Lattice dispatch id of the workflow to be triggered
+        dispatcher_addr: Address of the dispatcher server
+        triggers_server_addr: Address of the triggers server
+
+    Attributes:
+        self.db_path: Absolute path to the database file
+        self.table_name: Name of the table to observe
+        self.poll_interval: Time in seconds to wait for before reading the database again
+        self.where_clauses: List of "WHERE" conditions, e.g. ["id > 2", "status = pending"], to check when
+                            polling the database
+        self.trigger_after_n: Number of times the event must happen after which the workflow will be triggered.
+                              e.g value of 2 means workflow will be triggered once the event has occurred twice.
+        self.stop_flag: Thread safe flag used to check whether the stop condition has been met
+
+    """
+
+    def __init__(
+        self,
+        db_path: str,
+        table_name: str,
+        poll_interval: int = 1,
+        where_clauses: List[str] = None,
+        trigger_after_n: int = 1,
+        lattice_dispatch_id: str = None,
+        dispatcher_addr: str = None,
+        triggers_server_addr: str = None,
+    ):
+        super().__init__(lattice_dispatch_id, dispatcher_addr, triggers_server_addr)
+
+        self.db_path = db_path
+        self.table_name = table_name
+        self.poll_interval = poll_interval
+
+        self.where_clauses = where_clauses
+
+        self.trigger_after_n = trigger_after_n
+
+        self.stop_flag = None
+
+    def observe(self) -> None:
+        """
+        Keep performing the trigger action as long as
+        where conditions are met or until stop has being called
+        """
+
+        app_log.debug("Inside SQLiteTrigger's observe")
+        event_count = 0
+
+        connection = sqlite3.connect(self.db_path)
+        connection.row_factory = sqlite3.Row
+
+        cursor = connection.cursor()
+
+        sql_poll_cmd = f"SELECT * FROM {self.table_name}"
+
+        if self.where_clauses:
+            sql_poll_cmd += " WHERE "
+            sql_poll_cmd += " AND ".join(list(self.where_clauses))
+
+        execute_cmd = partial(cursor.execute, sql_poll_cmd)
+
+        self.stop_flag = Event()
+        while not self.stop_flag.is_set():
+            # Read the DB with specified command
+            try:
+                execute_cmd()
+            except sqlite3.OperationalError:
+                time.sleep(self.poll_interval)
+                continue
+
+            # If command ran successfuly, trigger the workflow
+            if selected_rows := cursor.fetchall():
+                event_count += 1
+                if event_count == self.trigger_after_n:
+                    self.trigger()
+                    event_count = 0
+
+            time.sleep(self.poll_interval)
+
+        cursor.close()
+        connection.close()
+
+    def stop(self) -> None:
+        """
+        Stop the running `self.observe()` method by setting the `self.stop_flag` flag.
+        """
+
+        self.stop_flag.set()
diff --git a/covalent/triggers/trigger_loader.py b/covalent/triggers/trigger_loader.py
index 91acb86b..6a5262cb 100644
--- a/covalent/triggers/trigger_loader.py
+++ b/covalent/triggers/trigger_loader.py
@@ -19,7 +19,7 @@
 # Relief from the License may be granted by purchasing a commercial license.
 
 
-from covalent.triggers import BaseTrigger, DirTrigger, TimeTrigger
+from covalent.triggers import BaseTrigger, DirTrigger, SQLiteTrigger, TimeTrigger
 
 
 class TriggerLoader:
@@ -28,6 +28,7 @@ class TriggerLoader:
             BaseTrigger.__name__: BaseTrigger,
             DirTrigger.__name__: DirTrigger,
             TimeTrigger.__name__: TimeTrigger,
+            SQLiteTrigger.__name__: SQLiteTrigger,
         }
 
     def __new__(cls):


Test patch:
diff --git a/tests/covalent_tests/triggers/sqlite_trigger_test.py b/tests/covalent_tests/triggers/sqlite_trigger_test.py
new file mode 100644
index 00000000..2b9e38ff
--- /dev/null
+++ b/tests/covalent_tests/triggers/sqlite_trigger_test.py
@@ -0,0 +1,135 @@
+# Copyright 2023 Agnostiq Inc.
+#
+# This file is part of Covalent.
+#
+# Licensed under the GNU Affero General Public License 3.0 (the "License").
+# A copy of the License may be obtained with this software package or at
+#
+#      https://www.gnu.org/licenses/agpl-3.0.en.html
+#
+# Use of this file is prohibited except in compliance with the License. Any
+# modifications or derivative works of this file must retain this copyright
+# notice, and modified files must contain a notice indicating that they have
+# been altered from the originals.
+#
+# Covalent is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the License for more details.
+#
+# Relief from the License may be granted by purchasing a commercial license.
+
+
+from sqlite3 import OperationalError
+
+import pytest
+
+from covalent.triggers.sqlite_trigger import SQLiteTrigger
+
+
+@pytest.fixture
+def sqlite_trigger():
+    """
+    Fixture to obtain a sample SQLiteTrigger instance for testing
+    """
+
+    db_path = "test_db_path"
+    table_name = "test_table_name"
+    poll_interval = 1
+    where_clauses = ["id > 2", "status = pending"]
+    trigger_after_n = 1
+    return SQLiteTrigger(db_path, table_name, poll_interval, where_clauses, trigger_after_n)
+
+
+def test_sqlite_trigger_init(sqlite_trigger):
+    """
+    Testing whether SQLiteTrigger initializes as expected
+    """
+
+    assert sqlite_trigger.db_path == "test_db_path"
+    assert sqlite_trigger.table_name == "test_table_name"
+    assert sqlite_trigger.poll_interval == 1
+    assert sqlite_trigger.where_clauses == ["id > 2", "status = pending"]
+    assert sqlite_trigger.trigger_after_n == 1
+
+
+@pytest.mark.parametrize(
+    "where_clauses",
+    [
+        ["id > 2", "status = pending"],
+        None,
+    ],
+)
+def test_sqlite_trigger_observe(mocker, where_clauses, sqlite_trigger):
+    """
+    Test the observe method of SQLiteTrigger
+    """
+
+    sqlite_trigger.where_clauses = where_clauses
+    sqlite_trigger.trigger = mocker.MagicMock()
+
+    mock_sqlite = mocker.patch("covalent.triggers.sqlite_trigger.sqlite3")
+    mock_event = mocker.patch("covalent.triggers.sqlite_trigger.Event")
+
+    mock_sleep = mocker.patch("covalent.triggers.sqlite_trigger.time.sleep")
+
+    mock_event.return_value.is_set.side_effect = [False, True]
+
+    sqlite_trigger.observe()
+
+    mock_sqlite.connect.assert_called_once_with("test_db_path")
+    mock_sqlite.connect.return_value.row_factory = mock_sqlite.Row
+    mock_sqlite.connect.return_value.cursor.assert_called_once()
+
+    mock_event.assert_called_once()
+
+    sql_poll_cmd = "SELECT * FROM test_table_name"
+    if where_clauses:
+        sql_poll_cmd += " WHERE "
+        sql_poll_cmd += " AND ".join(list(where_clauses))
+
+    mock_sqlite.connect.return_value.cursor.return_value.execute.assert_called_once_with(
+        sql_poll_cmd
+    )
+
+    mock_sqlite.connect.return_value.cursor.return_value.fetchall.assert_called_once()
+
+    sqlite_trigger.trigger.assert_called_once()
+    mock_sleep.assert_called_once_with(1)
+
+    mock_sqlite.connect.return_value.cursor.return_value.close.assert_called_once()
+    mock_sqlite.connect.return_value.close.assert_called_once()
+
+
+def test_sqlite_trigger_exception(mocker, sqlite_trigger):
+    """
+    Test the observe method of SQLiteTrigger when an OperationalError is raised
+    """
+
+    sqlite_trigger.trigger = mocker.MagicMock()
+
+    mock_connect = mocker.patch("covalent.triggers.sqlite_trigger.sqlite3.connect")
+    mocker.patch("covalent.triggers.sqlite_trigger.sqlite3.Row")
+    mock_event = mocker.patch("covalent.triggers.sqlite_trigger.Event")
+
+    mock_sleep = mocker.patch("covalent.triggers.sqlite_trigger.time.sleep")
+
+    mock_event.return_value.is_set.side_effect = [False, True]
+
+    mock_connect.return_value.cursor.return_value.execute.side_effect = OperationalError
+
+    sqlite_trigger.observe()
+
+    mock_sleep.assert_called_once_with(1)
+
+
+def test_sqlite_trigger_stop(mocker, sqlite_trigger):
+    """
+    Test the stop method of SQLiteTrigger
+    """
+
+    mock_stop_flag = mocker.MagicMock()
+    sqlite_trigger.stop_flag = mock_stop_flag
+
+    sqlite_trigger.stop()
+
+    mock_stop_flag.set.assert_called_once()