Compare commits

..

2 Commits

Author SHA1 Message Date
AcerecA f82086b2e5 remove use of pydantic 2026-01-02 16:11:38 +01:00
AcerecA 615b696fda replace use of pydantic_settings w/ dataclass 2026-01-02 16:11:31 +01:00
2 changed files with 93 additions and 38 deletions

View File

@ -5,7 +5,6 @@ description = "Add your description here"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"pydantic-settings~=2.12",
"pynvim~=0.6.0",
"requests>=2.32.5",
]

View File

@ -1,12 +1,13 @@
from dataclasses import dataclass, field
from collections.abc import Sequence
from dataclasses import dataclass, field, fields
from json import loads
from os import environ
from typing import Any, Callable
from pydantic import AliasPath, BaseModel, Field, field_validator
from typing import Any, Callable, Self, TypedDict
from urllib.request import Request, urlopen
from .hints import JSONDataMap
def get(
url: str,
@ -41,57 +42,84 @@ def put(
return resp.read().decode("utf-8")
class ClickupTask(BaseModel):
"""fields marked with `exclude=True` will not be pushed for updates"""
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 = Field(validation_alias=AliasPath("status", "status"))
status: str
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)
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)
@field_validator("checklists", mode="before")
@classmethod
def convert_checklists(cls, content: list[Any]) -> dict[str, bool]:
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"]
}
@field_validator("assignees", mode="before")
@classmethod
def convert_assignees(cls, content: list[str | dict[str, Any]]) -> list[str]:
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]
@field_validator("tags", "locations", mode="before")
@classmethod
def convert_simple_list_with_names(
cls,
content: list[str | dict[str, Any]],
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 {
k: getattr(self, k, None)
for k, v in type(self).model_fields.items()
if not v.exclude
f.name: getattr(self, f.name, None) for f in fields(type(self)) if f.repr
}
@property
def showables(self) -> dict[str, Any]:
return {
k: getattr(self, k, None)
for k in type(self).model_fields
if k != "markdown_description"
f.name: getattr(self, f.name, None)
for f in fields(type(self))
if f.name != "markdown_description"
}
@property
@ -106,18 +134,40 @@ class ClickupTask(BaseModel):
return ret
def get_env_var(var_name: str) -> Callable[[], str]:
return lambda var=var_name: environ[var]
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"))
workspace_id: str = field(default_factory=get_env_var("CLICKUP_WORKSPACE_ID"))
user_id: str = field(default_factory=get_env_var("CLICKUP_USER_ID"))
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: Any) -> dict[str, Any]:
def _get(self, endpoint: str, **query_params: str) -> JSONDataMap:
raw_data = get(
self.base_url + endpoint,
query_params,
@ -128,7 +178,7 @@ class ClickupSession:
)
return loads(raw_data)
def _put(self, endpoint: str, **body_params: Any) -> dict[str, Any]:
def _put(self, endpoint: str, **body_params: str) -> JSONDataMap:
raw_data = put(
self.base_url + endpoint,
body_params,
@ -143,12 +193,18 @@ class ClickupSession:
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"]]
).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.model_validate(
self._get(f"/task/{task_id}?include_markdown_description=true")
return ClickupTask.from_resp_data(
self._get(
f"/task/{task_id}",
include_markdown_description="true",
),
)
def update(self, data: ClickupTask) -> None: