hamster-tools/hamstertools/kimaiapi.py

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,
            },
        )