Spaces:
Sleeping
Sleeping
File size: 3,594 Bytes
4b9d59b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """
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"
# OAuth credentials from environment
self.client_id = os.getenv("CLICKUP_CLIENT_ID")
self.client_secret = os.getenv("CLICKUP_CLIENT_SECRET")
# In a real scenario, this would be fetched from UserConnection
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}")
|