From f82086b2e553b2d63097e13f13f04e14a04cde9e Mon Sep 17 00:00:00 2001 From: AcerecA Date: Fri, 2 Jan 2026 16:11:38 +0100 Subject: [PATCH] remove use of pydantic --- rplugin/python3/mrpy/clickup.py | 130 +++++++++++++++++++++++--------- 1 file changed, 93 insertions(+), 37 deletions(-) diff --git a/rplugin/python3/mrpy/clickup.py b/rplugin/python3/mrpy/clickup.py index a60a4f3..cae45bd 100644 --- a/rplugin/python3/mrpy/clickup.py +++ b/rplugin/python3/mrpy/clickup.py @@ -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: