replace use of pydantic_settings w/ dataclass

pyproject.toml

@@ -5,6 +5,7 @@ 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",
 ]

rplugin/python3/mrpy/clickup.py

@@ -1,13 +1,12 @@
-from collections.abc import Sequence
-from dataclasses import dataclass, field, fields
+from dataclasses import dataclass, field
 from json import loads
 from os import environ
-from typing import Any, Callable, Self, TypedDict
+from typing import Any, Callable
+
+from pydantic import AliasPath, BaseModel, Field, field_validator
 
 from urllib.request import Request, urlopen
 
-from .hints import JSONDataMap
-
 
 def get(
     url: str,
@@ -42,84 +41,57 @@ def put(
         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"""
+class ClickupTask(BaseModel):
+    """fields marked with `exclude=True` will not be pushed for updates"""
 
     name: str
     markdown_description: str
-    status: str
+    status: str = Field(validation_alias=AliasPath("status", "status"))
 
-    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)
+    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 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]:
+    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: Sequence[str | dict[str, str]]) -> list[str]:
+    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: Sequence[str | dict[str, str]],
+        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 {
-            f.name: getattr(self, f.name, None) for f in fields(type(self)) if f.repr
+            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 {
-            f.name: getattr(self, f.name, None)
-            for f in fields(type(self))
-            if f.name != "markdown_description"
+            k: getattr(self, k, None)
+            for k in type(self).model_fields
+            if k != "markdown_description"
         }
 
     @property
@@ -134,40 +106,18 @@ class ClickupTask:
         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
+def get_env_var(var_name: str) -> Callable[[], str]:
+    return lambda var=var_name: environ[var]
 
 
 @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",
-        )
-    )
+    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"))
     base_url: str = "https://api.clickup.com/api/v2"
 
-    def _get(self, endpoint: str, **query_params: str) -> JSONDataMap:
+    def _get(self, endpoint: str, **query_params: Any) -> dict[str, Any]:
         raw_data = get(
             self.base_url + endpoint,
             query_params,
@@ -178,7 +128,7 @@ class ClickupSession:
         )
         return loads(raw_data)
 
-    def _put(self, endpoint: str, **body_params: str) -> JSONDataMap:
+    def _put(self, endpoint: str, **body_params: Any) -> dict[str, Any]:
         raw_data = put(
             self.base_url + endpoint,
             body_params,
@@ -193,18 +143,12 @@ 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}"
-        ).get("tasks", [])
-        if isinstance(data, list):
-            return [ClickupTask.from_resp_data(t) for t in data if isinstance(t, dict)]
-
-        return []
+        )
+        return [ClickupTask.model_validate(t) for t in data["tasks"]]
 
     def get_task(self, task_id: str) -> ClickupTask:
-        return ClickupTask.from_resp_data(
-            self._get(
-                f"/task/{task_id}",
-                include_markdown_description="true",
-            ),
+        return ClickupTask.model_validate(
+            self._get(f"/task/{task_id}?include_markdown_description=true")
         )
 
     def update(self, data: ClickupTask) -> None: