Reasonably-working Kimai API data fetch'n'display

hamstertools/__init__.py

@@ -704,7 +704,6 @@ def _import(username, mapping_path=None, output=None, category_search=None, afte
 @cli.command()
 def app():
     from .app import HamsterToolsApp
-    #app = HamsterToolsApp(db_cursor=c, db_connection=conn)
     app = HamsterToolsApp()
     app.run()
 

hamstertools/app.py

@@ -5,7 +5,8 @@ from textual.containers import Horizontal, Vertical
 from textual.coordinate import Coordinate
 from textual.screen import Screen
 
-from .db import DatabaseManager, Category, Activity
+from .db import DatabaseManager, Category, Activity, KimaiProject, KimaiCustomer
+from .kimai import KimaiAPI, Customer as KimaiAPICustomer, Project as KimaiAPIProject, Activity as KimaiAPIActivity
 
 
 class ListScreen(Screen):
@@ -55,7 +56,6 @@ class ListScreen(Screen):
 
 class ActivitiesScreen(ListScreen):
     BINDINGS = [
-        ("q", "quit", "Quit"),
         ("s", "sort", "Sort"),
         ("r", "refresh", "Refresh"),
         ("/", "filter", "Search"),
@@ -68,7 +68,7 @@ class ActivitiesScreen(ListScreen):
         self.table.clear()
 
         # List activities with the count of facts
-        activities = Activity.list_activities(self.db_manager, filter_query)
+        activities = Activity.list(self.db_manager, filter_query)
 
         self.table.add_rows(
             [
@@ -133,7 +133,6 @@ class ActivitiesScreen(ListScreen):
 
 class CategoriesScreen(ListScreen):
     BINDINGS = [
-        ("q", "quit", "Quit"),
         ("s", "sort", "Sort"),
         ("r", "refresh", "Refresh"),
         ("/", "filter", "Search"),
@@ -144,7 +143,7 @@ class CategoriesScreen(ListScreen):
     def _refresh(self, filter_query=None):
         self.table.clear()
 
-        categories = Category.list_categories(
+        categories = Category.list(
             self.db_manager, filter_query=filter_query
         )
 
@@ -182,11 +181,66 @@ class CategoriesScreen(ListScreen):
         self.table.remove_row(row_key)
 
 
+class KimaiScreen(ListScreen):
+    BINDINGS = [
+        ("s", "sort", "Sort"),
+        ("r", "refresh", "Refresh"),
+        ("g", "get", "Get data"),
+        ("/", "filter", "Search"),
+        Binding(key="escape", action="cancelfilter", show=False),
+    ]
+
+    def _refresh(self, filter_query=None):
+        self.table.clear()
+
+        projects = KimaiProject.list(
+            self.db_manager
+        )
+
+        self.table.add_rows(
+            [
+                [
+                    project.customer_id,
+                    project.customer_name,
+                    project.id,
+                    project.name,
+                ]
+                for project in projects
+            ]
+        )
+
+        self.table.sort(self.sort)
+
+    def action_get(self) -> None:
+        api = KimaiAPI()
+
+        customers = KimaiAPICustomer.list(api)
+        for customer in customers:
+            KimaiCustomer(self.db_manager, id=customer.id, name=customer.name).save()
+
+        projects = KimaiAPIProject.list(api)
+        for project in projects:
+            KimaiProject(self.db_manager, id=project.id, name=project.name,
+                         customer_id=project.customer.id, customer_name="").save()
+
+        self._refresh()
+
+    def on_mount(self) -> None:
+        self.table = self.query_one(DataTable)
+        self.table.cursor_type = "row"
+        self.columns = self.table.add_columns("customer id", "customer",
+                                              "project id", "project")
+        # self.sort = (self.columns[1], self.columns[3])
+        self.sort = self.columns[1]
+        self._refresh()
+
+
 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"),
     ]
 
@@ -195,7 +249,8 @@ class HamsterToolsApp(App):
 
         self.MODES = {
             "categories": CategoriesScreen(self.db_manager),
-            "activities": ActivitiesScreen(self.db_manager)
+            "activities": ActivitiesScreen(self.db_manager),
+            "kimai": KimaiScreen(self.db_manager)
         }
 
         super().__init__()

hamstertools/db.py

@@ -1,5 +1,6 @@
 import sqlite3
 
+
 class DatabaseManager:
     def __init__(self, database_name):
         self.conn = sqlite3.connect(database_name)
@@ -14,6 +15,7 @@ class DatabaseManager:
     def close(self):
         self.conn.close()
 
+
 class BaseORM:
     def __init__(self, db_manager, table_name, id, **kwargs):
         self.db_manager = db_manager
@@ -31,11 +33,12 @@ class BaseORM:
 
 class Category(BaseORM):
     def __init__(self, db_manager, id, name, activity_count):
-        super().__init__(db_manager, "categories", id, name=name,
-                         activity_count=activity_count)
+        super().__init__(
+            db_manager, "categories", id, name=name, activity_count=activity_count
+        )
 
     @staticmethod
-    def list_categories(db_manager, filter_query=None):
+    def list(db_manager, filter_query=None):
         cursor = db_manager.get_cursor()
         where = ""
         if filter_query is not None:
@@ -65,7 +68,8 @@ class Category(BaseORM):
     @staticmethod
     def get_by_id(db_manager, category_id):
         cursor = db_manager.get_cursor()
-        cursor.execute("""
+        cursor.execute(
+            """
             SELECT
                 categories.id,
                 categories.name,
@@ -78,7 +82,9 @@ class Category(BaseORM):
                 categories.id = activities.category_id
             WHERE
                 categories.id = ?
-        """, (category_id,))
+        """,
+            (category_id,),
+        )
 
         row = cursor.fetchone()
         if row:
@@ -88,7 +94,9 @@ class Category(BaseORM):
 
 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)
+        super().__init__(
+            db_manager, "activities", id, name=name, category_id=category_id
+        )
         self.category_name = category_name
         self.facts_count = facts_count
 
@@ -97,19 +105,22 @@ class Activity(BaseORM):
 
         print(f"moving from {self.id} to {to_activity.id}")
 
-        cursor.execute("""
+        cursor.execute(
+            """
         UPDATE
             facts
         SET
             activity_id = ?
         WHERE
             activity_id = ?
-        """, (to_activity.id, self.id))
+        """,
+            (to_activity.id, self.id),
+        )
 
         self.conn.commit()
 
     @staticmethod
-    def list_activities(db_manager, filter_query=None):
+    def list(db_manager, filter_query=None):
         cursor = db_manager.get_cursor()
         where = ""
         if filter_query is not None:
@@ -142,12 +153,15 @@ class Activity(BaseORM):
             cursor.execute(sql)
 
         rows = cursor.fetchall()
-        return [Activity(db_manager, row[0], row[1], row[2], row[3], row[4]) for row in rows]
+        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("""
+        cursor.execute(
+            """
             SELECT
                 activities.id,
                 activities.name,
@@ -166,7 +180,9 @@ class Activity(BaseORM):
                 activities.id = facts.activity_id
             WHERE
                 activities.id = ?
-        """, (activity_id,))
+        """,
+            (activity_id,),
+        )
 
         row = cursor.fetchone()
         if row:
@@ -179,8 +195,74 @@ class Fact(BaseORM):
         super().__init__(db_manager, "facts", id, activity_id=activity_id)
 
     @staticmethod
-    def list_facts(db_manager):
+    def list(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]
+
+
+class KimaiCustomer(BaseORM):
+    def __init__(self, db_manager, id, name):
+        super().__init__(db_manager, "kimai_customers", id, name=name)
+
+    def save(self):
+        cursor = self.db_manager.get_cursor()
+        cursor.execute("SELECT id FROM kimai_customers WHERE id = ?", (self.id,))
+        row = cursor.fetchone()
+        if row:
+            cursor.execute("""
+                UPDATE kimai_customers SET name = ? WHERE id = ?
+            """, (self.name, self.id))
+        else:
+            cursor.execute("""
+                INSERT INTO kimai_customers (id, name) VALUES (?, ?)
+            """, (self.id, self.name))
+        self.db_manager.get_conn().commit()
+
+
+
+class KimaiProject(BaseORM):
+    def __init__(self, db_manager, id, name, customer_id, customer_name):
+        super().__init__(db_manager, "kimai_projects", id, name=name,
+                         customer_id=customer_id, customer_name=customer_name)
+
+    def save(self):
+        cursor = self.db_manager.get_cursor()
+        cursor.execute("SELECT id FROM kimai_projects WHERE id = ?", (self.id,))
+        row = cursor.fetchone()
+        if row:
+            cursor.execute("""
+                UPDATE kimai_projects SET name = ?, customer_id = ? WHERE id = ?
+            """, (self.name, self.customer_id, self.id))
+        else:
+            cursor.execute("""
+                INSERT INTO kimai_projects (id, name, customer_id) VALUES (?, ?, ?)
+            """, (self.id, self.name, self.customer_id))
+        self.db_manager.get_conn().commit()
+
+    @staticmethod
+    def list(db_manager):
+        cursor = db_manager.get_cursor()
+        cursor.execute("""
+            SELECT
+                kimai_projects.id,
+                COALESCE(kimai_projects.name, ""),
+                kimai_customers.id,
+                COALESCE(kimai_customers.name, "")
+            FROM
+                kimai_projects
+            LEFT JOIN
+                kimai_customers
+            ON
+                kimai_customers.id = kimai_projects.customer_id
+            GROUP BY
+                kimai_customers.id
+        """)
+
+        rows = cursor.fetchall()
+        return [KimaiProject(db_manager, row[0], row[1], row[2], row[3]) for row in rows]
+
+
+class KimaiACtivity(BaseORM):
+    pass

hamstertools/kimai.py

@@ -0,0 +1,80 @@
+import requests
+import requests_cache
+import os
+
+
+class NotFound(Exception):
+    pass
+
+
+class KimaiAPI(object):
+    # temporary hardcoded config
+    KIMAI_API_URL = 'https://kimai.autonomic.zone/api'
+
+    KIMAI_USERNAME = '3wordchant'
+    KIMAI_API_KEY = os.environ['KIMAI_API_KEY']
+
+    auth_headers = {
+        'X-AUTH-USER': KIMAI_USERNAME,
+        'X-AUTH-TOKEN': KIMAI_API_KEY
+    }
+
+    def __init__(self):
+        requests_cache.install_cache('kimai', backend='sqlite', expire_after=1800)
+        self.customers_json = requests.get(
+            f'{self.KIMAI_API_URL}/customers?visible=3', headers=self.auth_headers).json()
+        self.projects_json = requests.get(
+            f'{self.KIMAI_API_URL}/projects?visible=3', headers=self.auth_headers).json()
+
+
+class BaseAPI(object):
+    def __init__(self, api, **kwargs):
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+
+class Customer(BaseAPI):
+    def __init__(self, api, id, name):
+        super().__init__(api, id=id, name=name)
+
+    @staticmethod
+    def list(api):
+        return [
+            Customer(api, c['id'], c['name']) for c in api.customers_json
+        ]
+
+    @staticmethod
+    def get_by_id(api, id):
+        for value in api.customers_json:
+            if value['id'] == id:
+                return Customer(api, value['id'], value['name'])
+        raise NotFound()
+
+    def __repr__(self):
+        return f'Customer (id={self.id}, name={self.name})'
+
+
+class Project(BaseAPI):
+    def __init__(self, api, id, name, customer):
+        super().__init__(api, id=id, name=name, customer=customer)
+
+    @staticmethod
+    def list(api):
+        return [
+            Project(api, p['id'], p['name'], Customer.get_by_id(api, p['customer'])) for p in api.projects_json
+        ]
+
+    @staticmethod
+    def get_by_id(api, id):
+        for value in api.projects_json:
+            if value['id'] == id:
+                return Project(api, value['id'], value['name'],
+                               Customer.get_by_id(api, value['customer']))
+        raise NotFound()
+
+    def __repr__(self):
+        return f'Project (id={self.id}, name={self.name}, customer={self.customer})'
+
+
+class Activity():
+    pass