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 98 additions and 37 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,11 +1,13 @@
from collections.abc import Sequence
from dataclasses import dataclass, field, fields
from json import loads
from typing import Any
from pydantic import AliasPath, BaseModel, Field, field_validator
from pydantic_settings import BaseSettings
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,
@ -40,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
@ -105,13 +134,40 @@ class ClickupTask(BaseModel):
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=...)
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: Any) -> dict[str, Any]:
def _get(self, endpoint: str, **query_params: str) -> JSONDataMap:
raw_data = get(
self.base_url + endpoint,
query_params,
@ -122,7 +178,7 @@ class ClickupSession(BaseSettings):
)
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,
@ -137,12 +193,18 @@ class ClickupSession(BaseSettings):
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: