Add tinydb

11 files changed, 51 insertions, 25 deletions

diff --git a/.gitignore b/.gitignore
index 64777da..15eca83 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,2 +1,3 @@
 __pycache__/
 .DS_Store
+db.json
diff --git a/Aggrofile b/Aggrofile
index 429d214..a4bf93f 100644
--- a/Aggrofile
+++ b/Aggrofile
@@ -1,19 +1,17 @@
 {
+    "db_path": "db.json",
     "plugins": {
         "juusomikkonen_blog_rss": {
             "plugin": "FeedSourcePlugin",
-            "trigger_type": "schedule",
             "schedule_expr": "schedule.every(10).seconds",
             "feed_url": "https://juusomikkonen.com/feed.xml"
         },
         "title_filter": {
             "plugin": "FilterPlugin",
-            "trigger_type": "input_change",
             "filter_expr": "'TypeScript' in item.title"
         },
         "output_feed": {
             "plugin": "FeedSinkPlugin",
-            "trigger_type": "input_change",
             "feed_name": "test_feed"
         }
     },
diff --git a/aggro b/aggro
index 527de23..edf6763 100755
--- a/aggro
+++ b/aggro
@@ -1,3 +1,3 @@
 #!/bin/bash
 
-python src/aggro.py
+python aggro.py
diff --git a/aggro.py b/aggro.py
index b5c6ce0..30b176e 100644
--- a/aggro.py
+++ b/aggro.py
@@ -5,6 +5,7 @@ import time
 from app.AggroConfig import AggroConfig
 from app.MemoryState import memory_state
 from app.PluginManager import PluginManager
+from app.database import setup_db
 
 
 def run_plugin_thread():
@@ -22,9 +23,13 @@ if __name__ == "__main__":
         aggrofile_content = json.loads(f.read())
 
     aggro_config = AggroConfig(
-        plugins=aggrofile_content["plugins"], graph=aggrofile_content["graph"]
+        db_path=aggrofile_content.get("db_path", "db.json"),
+        plugins=aggrofile_content["plugins"],
+        graph=aggrofile_content["graph"],
     )
 
+    setup_db(aggro_config)
+
     manager = PluginManager(aggro_config)
     manager.build_plugin_instances()
 
diff --git a/app/AggroConfig.py b/app/AggroConfig.py
index e933eec..6ff6801 100644
--- a/app/AggroConfig.py
+++ b/app/AggroConfig.py
@@ -3,5 +3,6 @@ from dataclasses import dataclass
 
 @dataclass
 class AggroConfig:
+    db_path: str
     plugins: dict[str, dict[str, str]]
     graph: dict[str, list[str]]
diff --git a/app/FeedSinkPlugin.py b/app/FeedSinkPlugin.py
index c35a811..6b0125d 100644
--- a/app/FeedSinkPlugin.py
+++ b/app/FeedSinkPlugin.py
@@ -54,8 +54,11 @@ class Plugin(PluginInterface):
 
         return ET.tostring(rss, "unicode")
 
-    def process(self, items: list[Item]) -> list[Item]:
+    def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
         print(f"[FeedSinkPlugin#{self.id}] process called, n={len(items)}")
+        if source_id is None:
+            raise Exception(f"FeedSinkPlugin#{self.id} can not be scheduled")
+
         ret = self.build_xml(items)
         print(f"[FeedSinkPlugin#{self.id}] process returning XML:")
         print(ret)
diff --git a/app/FeedSourcePlugin.py b/app/FeedSourcePlugin.py
index 5453f80..7690c8d 100644
--- a/app/FeedSourcePlugin.py
+++ b/app/FeedSourcePlugin.py
@@ -12,8 +12,13 @@ class Plugin(PluginInterface):
 
         print(f"[FeedSourcePlugin#{self.id}] initialized")
 
-    def process(self, items: list[Item]) -> list[Item]:
+    def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
         print(f"[FeedSourcePlugin#{self.id}] process called")
+        if source_id is not None:
+            raise Exception(
+                f"FeedSourcePlugin#{self.id} can only be scheduled, trying to process items from source {source_id}"
+            )
+
         feed: Any = feedparser.parse(self.feed_url)  # type: ignore
         result_items: list[Item] = []
         if "bozo" in feed and feed["bozo"] == 1:
@@ -26,5 +31,7 @@ class Plugin(PluginInterface):
                 Item(title=d["title"], link=d["link"], description=d["description"])
             )
 
-        print(f"[FeedSourcePlugin#{self.id}] process returns items, n={len(items)}")
+        print(
+            f"[FeedSourcePlugin#{self.id}] process returns items, n={len(result_items)}"
+        )
         return result_items
diff --git a/app/FilterPlugin.py b/app/FilterPlugin.py
index 735540c..4e6ee3b 100644
--- a/app/FilterPlugin.py
+++ b/app/FilterPlugin.py
@@ -10,8 +10,11 @@ class Plugin(PluginInterface):
         self.filter_expr: str = get_param("filter_expr", params)
         print(f"[FilterPlugin#{self.id}] initialized")
 
-    def process(self, items: list[Item]) -> list[Item]:
+    def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
         print(f"[FilterPlugin#{self.id}] process called, n={len(items)}")
+        if source_id is None:
+            raise Exception(f"[FilterPlugin#{self.id}] can not be scheduled")
+
         expr = f"filter(lambda item: {self.filter_expr}, input_feed)"
         ret = list(eval(expr, {"input_feed": items}))
         print(f"[FilterPlugin#{self.id}] process returning items, n={len(ret)}")
diff --git a/app/PluginInterface.py b/app/PluginInterface.py
index f334034..6492ed1 100644
--- a/app/PluginInterface.py
+++ b/app/PluginInterface.py
@@ -10,5 +10,5 @@ class PluginInterface(ABC):
         self.params = params
 
     @abstractmethod
-    def process(self, items: list[Item]) -> list[Item]:
+    def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
         pass
diff --git a/app/PluginManager.py b/app/PluginManager.py
index 3ec9d44..da85428 100644
--- a/app/PluginManager.py
+++ b/app/PluginManager.py
@@ -21,8 +21,6 @@ class PluginManager:
         plugin_class = module.Plugin
         self._plugins[plugin_name] = plugin_class
 
-    # TODO instead of propagating procedurally,
-    # insert data into an next_node_id-identified input queue in tinydb
     def propagate(self, id: str, items: list[Item]):
         next_nodes: list[str]
         if id not in self.config.graph:
@@ -31,14 +29,14 @@ class PluginManager:
             next_nodes = self.config.graph[id]
 
         for next_node_id in next_nodes:
-            self.run_plugin_job(next_node_id, items)
+            self.run_plugin_job(next_node_id, id, items)
 
-    def run_plugin_job(self, id: str, items: list[Item] = []):
+    def run_plugin_job(self, id: str, source_id: str | None, items: list[Item] = []):
         if not memory_state.running:
             return
 
         plugin: PluginInterface = self.plugin_instances[id]
-        ret_items: list[Item] = plugin.process(items)
+        ret_items: list[Item] = plugin.process(source_id, items)
         self.propagate(id, ret_items)
 
     def build_plugin_instances(self):
@@ -47,20 +45,14 @@ class PluginManager:
             params: dict[str, str] = self.config.plugins[id]
 
             plugin_name = get_param("plugin", params)
-            trigger_type = get_param("trigger_type", params)
+            schedule_expr: str | None = params.get("schedule_expr", None)
 
             if plugin_name not in self._plugins:
                 self.load_plugin(plugin_name)
 
-            match trigger_type:
-                case "schedule":
-                    schedule_expr = get_param("schedule_expr", params)
-                    job: schedule.Job = eval(schedule_expr, {"schedule": schedule})
-                    job.do(self.run_plugin_job, id)  # type: ignore
-                case "input_change":
-                    pass
-                case _:
-                    raise Exception("unknown trigger_type: " + trigger_type)
+            if schedule_expr is not None:
+                job: schedule.Job = eval(schedule_expr, {"schedule": schedule})
+                job.do(self.run_plugin_job, id, None)  # type: ignore
 
             PluginClass: Any = self._plugins[plugin_name]
             plugin: PluginInterface = PluginClass(id=id, params=params)
diff --git a/app/database.py b/app/database.py
new file mode 100644
index 0000000..fdeac1a
--- /dev/null
+++ b/app/database.py
@@ -0,0 +1,16 @@
+from tinydb import TinyDB
+
+from app.AggroConfig import AggroConfig
+
+
+def setup_db(config: AggroConfig):
+    global database_manager
+    database_manager = DatabaseManager(config)
+
+
+class DatabaseManager:
+    def __init__(self, config: AggroConfig):
+        self.db = TinyDB(config.db_path)
+
+
+database_manager: DatabaseManager