| """ |
| ClickUp Integration Adapter |
| |
| Provides a unified interface for ClickUp services within the IntegrationFactory. |
| """ |
|
|
| import logging |
| import os |
| import httpx |
| from typing import Dict, Any, List, Optional |
| from datetime import datetime, timezone |
|
|
| logger = logging.getLogger(__name__) |
|
|
| class ClickUpAdapter: |
| """ |
| Adapter for ClickUp OAuth integration. |
| """ |
|
|
| def __init__(self, db=None, workspace_id: str = None): |
| self.db = db |
| self.workspace_id = workspace_id |
| self.service_name = "clickup" |
| self.base_url = "https://api.clickup.com/api/v2" |
| |
| |
| self.client_id = os.getenv("CLICKUP_CLIENT_ID") |
| self.client_secret = os.getenv("CLICKUP_CLIENT_SECRET") |
| |
| |
| self._access_token: Optional[str] = os.getenv("CLICKUP_ACCESS_TOKEN") |
|
|
| async def test_connection(self) -> bool: |
| """Test the ClickUp API connection""" |
| if not self._access_token: |
| return False |
| |
| try: |
| async with httpx.AsyncClient() as client: |
| response = await client.get( |
| f"{self.base_url}/user", |
| headers={"Authorization": self._access_token} |
| ) |
| return response.status_code == 200 |
| except Exception as e: |
| logger.error(f"ClickUp connection test failed: {e}") |
| return False |
|
|
| async def get_data(self, data_type: str, query: str = None, **kwargs) -> Dict[str, Any]: |
| """ |
| Generic method to fetch data from ClickUp. |
| |
| Supported data_types: |
| - teams: Get workspaces |
| - spaces: Get spaces in a team (requires team_id) |
| - tasks: Get tasks in a list (requires list_id) |
| - user: Get authorized user info |
| """ |
| if not self._access_token: |
| raise ValueError("ClickUp access token not configured") |
|
|
| async with httpx.AsyncClient() as client: |
| headers = {"Authorization": self._access_token} |
| |
| if data_type == "teams": |
| response = await client.get(f"{self.base_url}/team", headers=headers) |
| response.raise_for_status() |
| return {"ok": True, "data": response.json().get("teams", [])} |
| |
| elif data_type == "spaces": |
| team_id = query or kwargs.get("team_id") |
| if not team_id: |
| raise ValueError("team_id is required for spaces") |
| response = await client.get(f"{self.base_url}/team/{team_id}/space", headers=headers) |
| response.raise_for_status() |
| return {"ok": True, "data": response.json().get("spaces", [])} |
| |
| elif data_type == "tasks": |
| list_id = query or kwargs.get("list_id") |
| if not list_id: |
| raise ValueError("list_id is required for tasks") |
| response = await client.get(f"{self.base_url}/list/{list_id}/task", headers=headers) |
| response.raise_for_status() |
| return {"ok": True, "data": response.json().get("tasks", [])} |
| |
| elif data_type == "user": |
| response = await client.get(f"{self.base_url}/user", headers=headers) |
| response.raise_for_status() |
| return {"ok": True, "data": response.json().get("user")} |
| |
| else: |
| raise ValueError(f"Unsupported data type for ClickUp: {data_type}") |
|
|
