from collections.abc import Sequence from dataclasses import dataclass, field, fields from json import loads from os import environ from typing import Any, Callable, Self, TypedDict from urllib.error import HTTPError from .hints import JSONDataMap from .requests import get, put from .env import EnvVar 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 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) @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]: return { entry["name"]: entry["resolved"] for checklist in content for entry in checklist["items"] } @classmethod 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] @classmethod def convert_simple_list_with_names( cls, 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 { f.name: getattr(self, f.name, None) for f in fields(type(self)) if f.repr } @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" } @property def short(self) -> str: ret = "" if self.parent: ret += " \033[32m 󰙅 " ret += f"{self.name} (#{self.id})" return ret @dataclass class ClickupSession: auth_key: str = field( default_factory=EnvVar( "CLICKUP_AUTH", "clickup auth token is required to be set", ) ) workspace_id: str = field( default_factory=EnvVar( "CLICKUP_WORKSPACE_ID", "clickup workspace id is required to be set", ) ) user_id: str = field( default_factory=EnvVar( "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: str) -> JSONDataMap: try: raw_data = get( self.base_url + endpoint, query_params, headers={ "accept": "application/json", "Authorization": self.auth_key, }, ) return loads(raw_data) except HTTPError as e: e.add_note(e.url) raise def _put(self, endpoint: str, **body_params: str) -> JSONDataMap: 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}" ).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.from_resp_data( self._get( f"/task/{task_id}", include_markdown_description="true", ), ) def update(self, data: ClickupTask) -> None: _ = self._put(f"/task/{data.id}", **data.updateables)