150 lines
4.4 KiB
Python
150 lines
4.4 KiB
Python
from json import loads
|
|
from typing import Any
|
|
|
|
from pydantic import AliasPath, BaseModel, Field, field_validator
|
|
from pydantic_settings import BaseSettings
|
|
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
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 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]:
|
|
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: Any) -> dict[str, Any]:
|
|
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}"
|
|
)
|
|
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)
|