add files

rplugin/python3/mrpy/clickup.py

@@ -0,0 +1,114 @@
+from typing import Any
+from pydantic import AliasChoices, AliasPath, BaseModel, Field, field_validator
+from pydantic_settings import BaseSettings
+from requests import get, put
+from ruamel.yaml import Node, Representer, ScalarNode
+
+
+class ClickupTask(BaseModel):
+    """fields marked with `exclude=True` will not be pushed for updates"""
+
+    name: str
+    markdown_description: str
+    status: str = Field(validation_alias=AliasPath("status", "status"))
+
+    id: str = Field(exclude=True)
+    assignees: list[str] = Field(exclude=True)
+    tags: list[str] = Field(exclude=True)
+    parent: str | None = Field(None, exclude=True)
+    parent_list: str = Field(validation_alias=AliasPath("list", "id"), exclude=True)
+    locations: list[str] = Field(exclude=True)
+    checklists: dict[str, bool] = Field(exclude=True)
+
+    @field_validator("checklists", mode="before")
+    @classmethod
+    def convert_checklists(cls, content: list[Any]) -> dict[str, bool]:
+        return {
+            entry["name"]: entry["resolved"]
+            for checklist in content
+            for entry in checklist["items"]
+        }
+
+    @field_validator("assignees", mode="before")
+    @classmethod
+    def convert_assignees(cls, content: list[str | dict[str, Any]]) -> list[str]:
+        return [(e if isinstance(e, str) else e["username"]) for e in content]
+
+    @field_validator("tags", "locations", mode="before")
+    @classmethod
+    def convert_simple_list_with_names(
+        cls,
+        content: list[str | dict[str, Any]],
+    ) -> list[str]:
+        return [(e if isinstance(e, str) else e["name"]) for e in content]
+
+    @property
+    def updateables(self) -> dict[str, Any]:
+        return {
+            k: getattr(self, k, None)
+            for k, v in type(self).model_fields.items()
+            if not v.exclude
+        }
+
+    @property
+    def showables(self) -> dict[str, Any]:
+        return {
+            k: getattr(self, k, None)
+            for k in type(self).model_fields
+            if k != "markdown_description"
+        }
+
+    @property
+    def short(self) -> str:
+        ret = ""
+
+        if self.parent:
+            ret += " \033[32m 󰙅 "
+
+        ret += f"{self.name} (#{self.id})"
+
+        return ret
+
+
+class ClickupSession(BaseSettings):
+    auth_key: str = Field(alias="CLICKUP_AUTH", default=...)
+    workspace_id: str = Field(alias="CLICKUP_WORKSPACE_ID", default=...)
+    user_id: str = Field(alias="CLICKUP_USER_ID", default=...)
+    base_url: str = "https://api.clickup.com/api/v2"
+
+    def _get(self, endpoint: str, **query_params: Any) -> dict[str, Any]:
+        with get(
+            self.base_url + endpoint,
+            query_params,
+            headers={
+                "accept": "application/json",
+                "Authorization": self.auth_key,
+            },
+        ) as resp:
+            return resp.json()
+
+    def _put(self, endpoint: str, **body_params: Any) -> dict[str, Any]:
+        with put(
+            self.base_url + endpoint,
+            body_params,
+            headers={
+                "accept": "application/json",
+                "content-type": "application/json",
+                "Authorization": self.auth_key,
+            },
+        ) as resp:
+            return resp.json()
+
+    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}"
+        )
+        return [ClickupTask.model_validate(t) for t in data["tasks"]]
+
+    def get_task(self, task_id: str) -> ClickupTask:
+        return ClickupTask.model_validate(
+            self._get(f"/task/{task_id}?include_markdown_description=true")
+        )
+
+    def update(self, data: ClickupTask) -> None:
+        _ = self._put(f"/task/{data.id}", **data.updateables)