Add text-based search/filter for Conditions and Processes.

[plomtask] / plomtask / db.py

diff --git a/plomtask/db.py b/plomtask/db.py
index d2791b1bdf43da11d389b805fffd6b2791ae168d..548381e72e5c473047d7b90cb7f55ea1627b0ca6 100644 (file)
--- a/plomtask/db.py
+++ b/plomtask/db.py
@@ -185,6 +185,17 @@ class DatabaseConnection:
         return list(self.exec(f'SELECT * FROM {table_name} WHERE {key} = ?',
                               (target,)))
 
+    # def column_where_pattern(self,
+    #                          table_name: str,
+    #                          column: str,
+    #                          pattern: str,
+    #                          keys: list[str]) -> list[Any]:
+    #     """Return column of rows where one of keys matches pattern."""
+    #     targets = tuple([f'%{pattern}%'] * len(keys))
+    #     haystack = ' OR '.join([f'{k} LIKE ?' for k in keys])
+    #     sql = f'SELECT {column} FROM {table_name} WHERE {haystack}'
+    #     return [row[0] for row in self.exec(sql, targets)]
+
     def column_where(self, table_name: str, column: str, key: str,
                      target: int | str) -> list[Any]:
         """Return column of table where key == target."""
@@ -220,6 +231,7 @@ class BaseModel(Generic[BaseModelId]):
     to_save_relations: list[tuple[str, str, str]] = []
     id_: None | BaseModelId
     cache_: dict[BaseModelId, Self]
+    to_search: list[str] = []
 
     def __init__(self, id_: BaseModelId | None) -> None:
         if isinstance(id_, int) and id_ < 1:
@@ -342,6 +354,26 @@ class BaseModel(Generic[BaseModelId]):
                 items[item.id_] = item
         return list(items.values())
 
+    @classmethod
+    def matching(cls: type[BaseModelInstance], db_conn: DatabaseConnection,
+                 pattern: str) -> list[BaseModelInstance]:
+        """Return all objects whose .to_search match pattern."""
+        items = cls.all(db_conn)
+        if pattern:
+            filtered = []
+            for item in items:
+                for attr_name in cls.to_search:
+                    toks = attr_name.split('.')
+                    parent = item
+                    for tok in toks:
+                        attr = getattr(parent, tok)
+                        parent = attr
+                    if pattern in attr:
+                        filtered += [item]
+                        break
+            return filtered
+        return items
+
     def save(self, db_conn: DatabaseConnection) -> None:
         """Write self to DB and cache and ensure .id_.