Refucktoring

hamstertools/kimaiapi.py

@@ -0,0 +1,190 @@
+from datetime import datetime
+import requests
+import requests_cache
+import os
+
+from dataclasses import dataclass, field
+
+
+class NotFound(Exception):
+    pass
+
+
+class KimaiAPI(object):
+    # temporary hardcoded config
+    # KIMAI_API_URL = "https://kimai.autonomic.zone/api"
+    KIMAI_API_URL = "https://kimaitest.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 = self.get("customers", {"visible": 3})
+        self.projects_json = self.get("projects", {"visible": 3})
+        self.activities_json = self.get("activities", {"visible": 3})
+        self.user_json = self.get("users/me")
+
+    def get(self, endpoint, params=None):
+        return requests.get(
+            f"{self.KIMAI_API_URL}/{endpoint}", params=params, headers=self.auth_headers
+        ).json()
+
+    def post(self, endpoint, data):
+        return requests.post(
+            f"{self.KIMAI_API_URL}/{endpoint}", json=data, headers=self.auth_headers
+        )
+
+
+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})"
+
+
+@dataclass
+class Project(BaseAPI):
+    api: KimaiAPI = field(repr=False)
+    id: int
+    name: str
+    customer: Customer
+    allow_global_activities: bool = field(default=True)
+
+    @staticmethod
+    def list(api):
+        return [
+            Project(
+                api,
+                p["id"],
+                p["name"],
+                Customer.get_by_id(api, p["customer"]),
+                p["globalActivities"],
+            )
+            for p in api.projects_json
+        ]
+
+    @staticmethod
+    def get_by_id(api, id, none=False):
+        for value in api.projects_json:
+            if value["id"] == id:
+                return Project(
+                    api,
+                    value["id"],
+                    value["name"],
+                    Customer.get_by_id(api, value["customer"]),
+                    value["globalActivities"],
+                )
+        if not none:
+            raise NotFound()
+
+
+@dataclass
+class Activity(BaseAPI):
+    api: KimaiAPI = field(repr=False)
+    id: int
+    name: str
+    project: Project
+
+    @staticmethod
+    def list(api):
+        return [
+            Activity(
+                api, a["id"], a["name"], Project.get_by_id(api, a["project"], none=True)
+            )
+            for a in api.activities_json
+        ]
+
+    @staticmethod
+    def get_by_id(api, id, none=False):
+        for value in api.activities_json:
+            if value["id"] == id:
+                return Activity(
+                    api,
+                    value["id"],
+                    value["name"],
+                    Project.get_by_id(api, value["project"]),
+                )
+        if not none:
+            raise NotFound()
+
+
+@dataclass
+class Timesheet(BaseAPI):
+    api: KimaiAPI = field(repr=False)
+    activity: Activity
+    project: Project
+    begin: datetime
+    end: datetime
+    id: int = field(default=None)
+    description: str = field(default="")
+    tags: str = field(default="")
+
+    @staticmethod
+    def list(api):
+        return [
+            Timesheet(
+                api,
+                Activity.get_by_id(api, t["activity"], none=True),
+                Project.get_by_id(api, t["project"], none=True),
+                t["begin"],
+                t["end"],
+                t["id"],
+                t["description"],
+                t["tags"],
+            )
+            for t in api.get(
+                "timesheets",
+            )
+        ]
+
+    @staticmethod
+    def get_by_id(api, id, none=False):
+        t = api.get(
+            f"timesheets/{id}",
+        )
+        return Timesheet(
+            api,
+            Activity.get_by_id(api, t["activity"], none=True),
+            Project.get_by_id(api, t["project"], none=True),
+            t["begin"],
+            t["end"],
+            t["id"],
+            t["description"],
+            t["tags"],
+        )
+
+    def upload(self):
+        return self.api.post(
+            "timesheets",
+            {
+                "begin": self.begin.isoformat(),
+                "end": self.end.isoformat(),
+                "project": self.project.id,
+                "activity": self.activity.id,
+                "description": self.description,
+                # FIXME: support multiple users
+                # "user": self.,
+                "tags": self.tags,
+            },
+        )