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 urllib.request import Request, urlopen

from .hints import JSONDataMap


def get(
    url: str,
    query_params: dict[str, str] | None = None,
    headers: dict[str, str] | None = None,
) -> str:
    with urlopen(
        Request(
            url + ("?" + "&".join(f"{k}={v}" for k, v in query_params.items()))
            if query_params
            else "",
            headers=headers or {},
            method="GET",
        ),
    ) as resp:
        return resp.read().decode("utf-8")


def put(
    url: str,
    data: dict[str, str] | None = None,
    headers: dict[str, str] | None = None,
) -> str:
    with urlopen(
        Request(
            url,
            str(data).encode(),
            headers=headers or {},
            method="PUT",
        ),
    ) as resp:
        return resp.read().decode("utf-8")


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


def get_env_var(var_name: str, err_msg: str = "") -> Callable[[], str]:
    def wrapper(var: str = var_name) -> str:
        try:
            return environ[var]
        except KeyError as e:
            e.add_note(err_msg)
            raise

    return wrapper


@dataclass
class ClickupSession:
    auth_key: str = field(
        default_factory=get_env_var(
            "CLICKUP_AUTH",
            "clickup auth token is required to be set",
        )
    )
    workspace_id: str = field(
        default_factory=get_env_var(
            "CLICKUP_WORKSPACE_ID",
            "clickup workspace id is required to be set",
        )
    )
    user_id: str = field(
        default_factory=get_env_var(
            "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)