Proper DB ORM, filtering 🎉

hamstertools/__init__.py

@@ -9,8 +9,9 @@ import click
 import requests
 import sqlite3
 
-HAMSTER_DIR = Path.home() / '.local/share/hamster'
-HAMSTER_FILE = HAMSTER_DIR / 'hamster.db'
+# HAMSTER_DIR = Path.home() / '.local/share/hamster'
+# HAMSTER_FILE = HAMSTER_DIR / 'hamster.db'
+HAMSTER_FILE = 'hamster-testing.db'
 
 conn = sqlite3.connect(HAMSTER_FILE)
 c = conn.cursor()
@@ -700,6 +701,14 @@ def _import(username, mapping_path=None, output=None, category_search=None, afte
     output_file.close()
 
 
+@cli.command()
+def app():
+    from .app import HamsterToolsApp
+    #app = HamsterToolsApp(db_cursor=c, db_connection=conn)
+    app = HamsterToolsApp()
+    app.run()
+
+
 @cli.command()
 def hamster():
     click.echo('🐹')

hamstertools/app.py

@@ -0,0 +1,218 @@
+from textual.app import App, ComposeResult
+from textual.binding import Binding
+from textual.widgets import Header, Footer, DataTable, Input
+from textual.containers import Horizontal, Vertical
+from textual.coordinate import Coordinate
+from textual.screen import Screen
+from textual.reactive import reactive
+
+from .db import DatabaseManager, Category, Activity
+
+class ActivitiesScreen(Screen):
+    BINDINGS = [
+        ("q", "quit", "Quit"),
+        ("s", "sort", "Sort"),
+        ("r", "refresh", "Refresh"),
+        ("d", "delete", "Delete activity"),
+        ("/", "filter", "Search"),
+        Binding(key="escape", action="cancelfilter", show=False),
+    ]
+
+    def __init__(self, db_manager):
+        self.db_manager = db_manager
+        super().__init__()
+
+    def _refresh(self, filter_query=None):
+        self.table.clear()
+
+        # List activities with the count of facts
+        activities = Activity.list_activities(self.db_manager, filter_query)
+
+        self.table.add_rows([[
+            activity.category_id,
+            activity.category_name,
+            activity.id,
+            activity.name,
+            activity.facts_count,
+        ] for activity in activities])
+
+        self.table.sort(*self.sort)
+
+    def compose(self) -> ComposeResult:
+        """create child widgets for the app."""
+        yield Header()
+        with Vertical():
+            yield DataTable()
+            with Horizontal():
+                yield Input(id="filter")
+        yield Footer()
+
+    def on_mount(self) -> None:
+        self.table = self.query_one(DataTable)
+        self.table.cursor_type = "row"
+        self.columns = self.table.add_columns("category id","category","activity id","activity","entries")
+        self.sort = (self.columns[1], self.columns[3])
+        self._refresh()
+
+    def action_refresh(self) -> None:
+        self._refresh()
+
+    def action_sort(self) -> None:
+        self.table.cursor_type = "column"
+
+    def action_filter(self) -> None:
+        filter_input = self.query_one("#filter")
+        filter_input.display = True
+        filter_input.focus()
+        print(filter_input)
+
+    def action_cancelfilter(self) -> None:
+        filter_input = self.query_one("#filter")
+        filter_input.display = False
+        self._refresh()
+
+    def action_delete(self) -> None:
+         # get the keys for the row and column under the cursor.
+        row_key, _ = self.table.coordinate_to_cell_key(self.table.cursor_coordinate)
+
+        activity_id = self.table.get_cell_at(
+            Coordinate(self.table.cursor_coordinate.row, 2),
+        )
+
+        activity = Activity.get_by_id(self.db_manager, activity_id)
+        activity.delete()
+
+        # supply the row key to `remove_row` to delete the row.
+        self.table.remove_row(row_key)
+
+    def on_data_table_column_selected(self, event):
+        self.sort = (event.column_key,)
+        event.data_table.sort(*self.sort)
+        event.data_table.cursor_type = "row"
+
+    def on_input_changed(self, event):
+        self._refresh(event.value)
+
+
+class CategoriesScreen(Screen):
+    BINDINGS = [
+        ("s", "sort", "Sort"),
+        ("r", "refresh", "Refresh"),
+        ("d", "delete", "Delete category"),
+        ("/", "filter", "Search"),
+        Binding(key="escape", action="cancelfilter", show=False),
+    ]
+
+    filtering = reactive(False)
+    filter_query = reactive("")
+
+    def __init__(self, db_manager):
+        self.db_manager = db_manager
+        super().__init__()
+
+    def _refresh(self, filter_query=None):
+        self.table.clear()
+
+        categories = Category.list_categories(self.db_manager,
+                                              filter_query=filter_query)
+
+        self.table.add_rows([[
+            category.id,
+            category.name,
+            category.activity_count,
+        ] for category in categories])
+
+        self.table.sort(self.sort)
+
+    def compose(self) -> ComposeResult:
+        """create child widgets for the app."""
+        yield Header()
+        with Vertical():
+            yield DataTable()
+            with Horizontal():
+                yield Input(id="filter")
+        yield Footer()
+
+    def on_mount(self) -> None:
+        self.table = self.query_one(DataTable)
+        self.table.cursor_type = "row"
+        self.columns = self.table.add_columns("category id","category","activities")
+        self.sort = self.columns[1]
+        self._refresh()
+
+    def action_refresh(self) -> None:
+        self._refresh()
+
+    def action_sort(self) -> None:
+        self.table.cursor_type = "column"
+
+    def action_filter(self) -> None:
+        filter_input = self.query_one("#filter")
+        filter_input.display = True
+        filter_input.focus()
+        print(filter_input)
+
+    def action_cancelfilter(self) -> None:
+        filter_input = self.query_one("#filter")
+        filter_input.display = False
+        self._refresh()
+
+    def on_input_changed(self, event):
+        self._refresh(event.value)
+
+    def action_delete(self) -> None:
+        # get the keys for the row and column under the cursor.
+        row_key, _ = self.table.coordinate_to_cell_key(self.table.cursor_coordinate)
+
+        category_id = self.table.get_cell_at(
+            Coordinate(self.table.cursor_coordinate.row, 0),
+        )
+        category = Category.get_by_id(self.db_manager, category_id)
+        category.delete()
+
+        # supply the row key to `remove_row` to delete the row.
+        self.table.remove_row(row_key)
+
+    def on_data_table_column_selected(self, event):
+        """ Handle column selection for sort """
+        self.sort = event.column_key
+        event.data_table.sort(self.sort)
+        event.data_table.cursor_type = "row"
+
+# class KimaiScreen(Screen):
+#     def compose(self) -> ComposeResult:
+#         yield Placeholder("Help Screen")
+#         yield Footer()
+#
+#     def __init__(self, db_cursor, db_connection):
+#         self.db_cursor = db_cursor
+#         self.db_connection = db_connection
+#         super().__init__()
+
+
+class HamsterToolsApp(App):
+    CSS_PATH = 'app.tcss'
+    BINDINGS = [
+        ("a", "switch_mode('activities')", "Activities"),
+        ("c", "switch_mode('categories')", "Categories"),
+        # ("k", "switch_mode('kimai')", "Kimai"),
+        ("q", "quit", "Quit"),
+    ]
+
+    def __init__(self):
+        self.db_manager = DatabaseManager('hamster-testing.db')
+
+        self.MODES = {
+            "categories": CategoriesScreen(self.db_manager),
+            "activities": ActivitiesScreen(self.db_manager)
+            # "kimai": KimaiScreen,
+        }
+
+        super().__init__()
+
+    def on_mount(self) -> None:
+        self.switch_mode("activities")
+
+    def action_quit(self) -> None:
+        self.exit()
+        self.db_manager.close()

hamstertools/db.py

@@ -0,0 +1,170 @@
+import sqlite3
+
+class DatabaseManager:
+    def __init__(self, database_name):
+        self.conn = sqlite3.connect(database_name)
+        self.cursor = self.conn.cursor()
+
+    def get_conn(self):
+        return self.conn
+
+    def get_cursor(self):
+        return self.cursor
+
+    def close(self):
+        self.conn.close()
+
+class BaseORM:
+    def __init__(self, db_manager, table_name, id, **kwargs):
+        self.db_manager = db_manager
+        self.conn = db_manager.get_conn()
+        self.cursor = db_manager.get_cursor()
+        self.id = id
+        self.table_name = table_name
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+    def delete(self):
+        self.cursor.execute(f"DELETE FROM {self.table_name} WHERE id=?", (self.id,))
+        self.conn.commit()
+
+
+class Category(BaseORM):
+    def __init__(self, db_manager, id, name, activity_count):
+        super().__init__(db_manager, "categories", id, name=name,
+                         activity_count=activity_count)
+
+    @staticmethod
+    def list_categories(db_manager, filter_query=None):
+        cursor = db_manager.get_cursor()
+        where = ""
+        if filter_query is not None:
+            where = "WHERE categories.name LIKE ?"
+        sql = f"""
+            SELECT
+                categories.id,
+                COALESCE(categories.name, ""),
+                COUNT(activities.id) AS activity_count
+            FROM
+                categories
+            LEFT JOIN
+                activities
+            ON
+                categories.id = activities.category_id
+            {where}
+            GROUP BY
+                categories.id
+            """
+        if filter_query is not None:
+            cursor.execute(sql, ("%{}%".format(filter_query),))
+        else:
+            cursor.execute(sql)
+        rows = cursor.fetchall()
+        return [Category(db_manager, row[0], row[1], row[2]) for row in rows]
+
+    @staticmethod
+    def get_by_id(db_manager, category_id):
+        cursor = db_manager.get_cursor()
+        cursor.execute("""
+            SELECT
+                categories.id,
+                categories.name,
+                COUNT(activities.id) AS activity_count
+            FROM
+                categories
+            LEFT JOIN
+                activities
+            ON
+                categories.id = activities.category_id
+            WHERE
+                categories.id = ?
+        """, (category_id,))
+
+        row = cursor.fetchone()
+        if row:
+            return Category(db_manager, row[0], row[1], row[2])
+        return None
+
+
+class Activity(BaseORM):
+    def __init__(self, db_manager, id, name, category_id, category_name, facts_count):
+        super().__init__(db_manager, "activities", id, name=name, category_id=category_id)
+        self.category_name = category_name
+        self.facts_count = facts_count
+
+    @staticmethod
+    def list_activities(db_manager, filter_query=None):
+        cursor = db_manager.get_cursor()
+        where = ""
+        if filter_query is not None:
+            where = "WHERE categories.name LIKE ? or activities.name like ?"
+        sql = f"""
+            SELECT
+                activities.id,
+                activities.name,
+                categories.id,
+                COALESCE(categories.name, ""),
+                COUNT(facts.id) AS facts_count
+            FROM
+                activities
+            LEFT JOIN
+                categories
+            ON
+                activities.category_id = categories.id
+            LEFT JOIN
+                facts
+            ON
+                activities.id = facts.activity_id
+            {where}
+            GROUP BY
+                activities.id
+        """
+
+        if filter_query is not None:
+            cursor.execute(sql, ("%{}%".format(filter_query),) * 2 )
+        else:
+            cursor.execute(sql)
+
+        rows = cursor.fetchall()
+        return [Activity(db_manager, row[0], row[1], row[2], row[3], row[4]) for row in rows]
+
+    @staticmethod
+    def get_by_id(db_manager, activity_id):
+        cursor = db_manager.get_cursor()
+        cursor.execute("""
+            SELECT
+                activities.id,
+                activities.name,
+                categories.id,
+                COALESCE(categories.name, ""),
+                COUNT(facts.id) AS facts_count
+            FROM
+                activities
+            LEFT JOIN
+                categories
+            ON
+                activities.category_id = categories.id
+            LEFT JOIN
+                facts
+            ON
+                activities.id = facts.activity_id
+            WHERE
+                activities.id = ?
+        """, (activity_id,))
+
+        row = cursor.fetchone()
+        if row:
+            return Activity(db_manager, row[0], row[1], row[2], row[3], row[4])
+        return None
+
+
+class Fact(BaseORM):
+    def __init__(self, db_manager, id, activity_id):
+        super().__init__(db_manager, "facts", id, activity_id=activity_id)
+
+    @staticmethod
+    def list_facts(db_manager):
+        cursor = db_manager.get_cursor()
+        cursor.execute("SELECT * FROM facts")
+        rows = cursor.fetchall()
+        return [Fact(db_manager, row[0], row[1]) for row in rows]