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, "ignoreDates": 1}) 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, }, )