from dataclasses import dataclass, field from json import loads from pydantic import BaseModel from requests import HTTPError, get from .common import EnvVar, JSONDataMap, JSONData, JSONDataList, JSONDataScalar class ClickupTask(BaseModel): id: str name: str @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: with get( self.base_url + endpoint, query_params, headers={ "accept": "application/json", "Authorization": self.auth_key, }, ) as resp: return resp.json() def get_tasks(self, **filters: str) -> list[ClickupTask]: data = self._get( f"/team/{self.workspace_id}/task", **{ "subtasks": "true", "include_markdown_description": "true", "assignees[]": self.user_id, } | filters, ).get("tasks", []) if isinstance(data, list): return [ClickupTask.model_validate(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", ), )