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): result = requests.get( f"{self.KIMAI_API_URL}/{endpoint}", params=params, headers=self.auth_headers ).json() try: if result["code"] != 200: raise NotFound() except (KeyError, TypeError): pass return result 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) @dataclass class Customer(BaseAPI): api: KimaiAPI = field(repr=False) id: int name: str visible: bool = field(default=True) @staticmethod def list(api): return [ Customer(api, c["id"], c["name"], c["visible"]) for c in api.customers_json ] @staticmethod def get_by_id(api, id): for c in api.customers_json: if c["id"] == id: return Customer(api, c["id"], c["name"], c["visible"]) raise NotFound() @dataclass class Project(BaseAPI): api: KimaiAPI = field(repr=False) id: int name: str customer: Customer allow_global_activities: bool = field(default=True) visible: bool = field(default=True) @staticmethod def list(api): return [ Project( api, p["id"], p["name"], Customer.get_by_id(api, p["customer"]), p["globalActivities"], p["visible"], ) for p in api.projects_json ] @staticmethod def get_by_id(api, id, none=False): for p in api.projects_json: if p["id"] == id: return Project( api, p["id"], p["name"], Customer.get_by_id(api, p["customer"]), p["globalActivities"], p["visible"], ) if not none: raise NotFound() @dataclass class Activity(BaseAPI): api: KimaiAPI = field(repr=False) id: int name: str project: Project visible: bool = field(default=True) @staticmethod def list(api): return [ Activity( api, a["id"], a["name"], Project.get_by_id(api, a["project"], none=True), a["visible"], ) for a in api.activities_json ] @staticmethod def get_by_id(api, id, none=False): for a in api.activities_json: if a["id"] == id: return Activity( api, a["id"], a["name"], Project.get_by_id(api, a["project"], none), a["visible"], ) 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 list_by(api, **kwargs): kwargs['size'] = 10000 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", params=kwargs ) ] @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, }, )