50 lines
2.2 KiB
Python
50 lines
2.2 KiB
Python
# Copyright (c) 2025 Patrick Motsch
|
|
# All rights reserved.
|
|
"""Resolve ClickUp UserConnection and configure ClickupService."""
|
|
|
|
import logging
|
|
from typing import Any, Dict, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class ClickupConnectionHelper:
|
|
def __init__(self, method_instance: Any):
|
|
self.method = method_instance
|
|
self.services = method_instance.services
|
|
|
|
def get_clickup_connection(self, connection_reference: str) -> Optional[Dict[str, Any]]:
|
|
try:
|
|
ref = (connection_reference or "").split(" [")[0].strip()
|
|
if not ref:
|
|
return None
|
|
user_connection = None
|
|
if ref.startswith("connection:"):
|
|
user_connection = self.services.chat.getUserConnectionFromConnectionReference(ref)
|
|
else:
|
|
app = getattr(self.services, "interfaceDbApp", None)
|
|
if app and hasattr(app, "getUserConnectionById"):
|
|
user_connection = app.getUserConnectionById(ref)
|
|
if not user_connection:
|
|
logger.warning("No user connection for reference/id %s", connection_reference)
|
|
return None
|
|
authority = getattr(user_connection.authority, "value", None) or str(
|
|
user_connection.authority
|
|
)
|
|
if authority != "clickup":
|
|
logger.warning("Connection %s is not ClickUp (authority=%s)", user_connection.id, authority)
|
|
return None
|
|
status = getattr(user_connection.status, "value", None) or str(user_connection.status)
|
|
if status not in ("active", "pending"):
|
|
logger.warning("Connection %s status not active: %s", user_connection.id, status)
|
|
|
|
cu = getattr(self.services, "clickup", None)
|
|
if not cu:
|
|
return None
|
|
if not cu.setAccessTokenFromConnection(user_connection):
|
|
logger.warning("Failed to set ClickUp token for connection %s", user_connection.id)
|
|
return None
|
|
return {"id": user_connection.id, "userConnection": user_connection}
|
|
except Exception as e:
|
|
logger.error("get_clickup_connection error: %s", e)
|
|
return None
|