mrpy.nvim/rplugin/python3/mrpy/clickup.py

from collections.abc import Sequence
from dataclasses import dataclass, field, fields
from json import loads
from os import environ
from typing import Any, Callable, Self, TypedDict

from .hints import JSONDataMap
from .requests import get, put
from .env import EnvVar


class TaskRespDict(TypedDict):
    name: str
    markdown_description: str
    status: dict[str, str]

    id: str
    assignees: list[dict[str, str]]
    tags: list[dict[str, str]]
    parent: str | None
    locations: list[dict[str, str]]
    checklists: dict[str, bool]
    list: dict[str, str]


@dataclass(kw_only=True)
class ClickupTask:
    """fields marked with ``repr=False`` will not be pushed for updates"""

    name: str
    markdown_description: str
    status: str

    id: str = field(repr=False)
    assignees: list[str] = field(repr=False)
    tags: list[str] = field(repr=False)
    parent_list: str = field(repr=False)
    locations: list[str] = field(repr=False)
    checklists: dict[str, bool] = field(repr=False)
    parent: str | None = field(default=None, repr=False)

    @classmethod
    def from_resp_data(cls, resp_data_raw: dict[str, Any]) -> Self:
        resp_data = TaskRespDict(**resp_data_raw)

        return cls(
            name=resp_data["name"],
            markdown_description=resp_data["markdown_description"],
            status=resp_data["status"]["status"],
            id=resp_data["id"],
            assignees=cls.convert_assignees(resp_data["assignees"]),
            tags=cls.convert_simple_list_with_names(resp_data["tags"]),
            parent=resp_data.get("parent"),
            parent_list=resp_data["list"]["id"],
            locations=cls.convert_simple_list_with_names(resp_data["locations"]),
            checklists=resp_data["checklists"],
        )

    @classmethod
    def convert_checklists(cls, content: Sequence[Any]) -> dict[str, bool]:
        return {
            entry["name"]: entry["resolved"]
            for checklist in content
            for entry in checklist["items"]
        }

    @classmethod
    def convert_assignees(cls, content: Sequence[str | dict[str, str]]) -> list[str]:
        return [(e if isinstance(e, str) else e["username"]) for e in content]

    @classmethod
    def convert_simple_list_with_names(
        cls,
        content: Sequence[str | dict[str, str]],
    ) -> list[str]:
        return [(e if isinstance(e, str) else e["name"]) for e in content]

    @property
    def updateables(self) -> dict[str, Any]:
        return {
            f.name: getattr(self, f.name, None) for f in fields(type(self)) if f.repr
        }

    @property
    def showables(self) -> dict[str, Any]:
        return {
            f.name: getattr(self, f.name, None)
            for f in fields(type(self))
            if f.name != "markdown_description"
        }

    @property
    def short(self) -> str:
        ret = ""

        if self.parent:
            ret += " \033[32m 󰙅 "

        ret += f"{self.name} (#{self.id})"

        return ret


@dataclass
class ClickupSession:
    auth_key: str = field(
        default_factory=EnvVar(
            "CLICKUP_AUTH",
            "clickup auth token is required to be set",
        )
    )
    workspace_id: str = field(
        default_factory=EnvVar(
            "CLICKUP_WORKSPACE_ID",
            "clickup workspace id is required to be set",
        )
    )
    user_id: str = field(
        default_factory=EnvVar(
            "CLICKUP_USER_ID",
            "clickup user id is required to be set",
        )
    )
    base_url: str = "https://api.clickup.com/api/v2"

    def _get(self, endpoint: str, **query_params: str) -> JSONDataMap:
        raw_data = get(
            self.base_url + endpoint,
            query_params,
            headers={
                "accept": "application/json",
                "Authorization": self.auth_key,
            },
        )
        return loads(raw_data)

    def _put(self, endpoint: str, **body_params: str) -> JSONDataMap:
        raw_data = put(
            self.base_url + endpoint,
            body_params,
            headers={
                "accept": "application/json",
                "content-type": "application/json",
                "Authorization": self.auth_key,
            },
        )
        return loads(raw_data)

    def get_tasks(self) -> list[ClickupTask]:
        data = self._get(
            f"/team/{self.workspace_id}/task?subtasks=true&include_markdown_description=true&assignees[]={self.user_id}"
        ).get("tasks", [])
        if isinstance(data, list):
            return [ClickupTask.from_resp_data(t) for t in data if isinstance(t, dict)]

        return []

    def get_task(self, task_id: str) -> ClickupTask:
        return ClickupTask.from_resp_data(
            self._get(
                f"/task/{task_id}",
                include_markdown_description="true",
            ),
        )

    def update(self, data: ClickupTask) -> None:
        _ = self._put(f"/task/{data.id}", **data.updateables)