Source code for langchain_community.utilities.clickup

"""Util that calls clickup."""

import json
import warnings
from dataclasses import asdict, dataclass, fields
from typing import Any, Dict, List, Mapping, Optional, Tuple, Type, Union

import requests
from langchain_core.utils import get_from_dict_or_env
from pydantic import BaseModel, ConfigDict, model_validator

DEFAULT_URL = "https://api.clickup.com/api/v2"



[docs]
@dataclass
class Component:
    """Base class for all components."""


[docs]
    @classmethod
    def from_data(cls, data: Dict[str, Any]) -> "Component":
        raise NotImplementedError()





[docs]
@dataclass
class Task(Component):
    """Class for a task."""

    id: int
    name: str
    text_content: str
    description: str
    status: str
    creator_id: int
    creator_username: str
    creator_email: str
    assignees: List[Dict[str, Any]]
    watchers: List[Dict[str, Any]]
    priority: Optional[str]
    due_date: Optional[str]
    start_date: Optional[str]
    points: int
    team_id: int
    project_id: int


[docs]
    @classmethod
    def from_data(cls, data: Dict[str, Any]) -> "Task":
        priority = None if data["priority"] is None else data["priority"]["priority"]
        return cls(
            id=data["id"],
            name=data["name"],
            text_content=data["text_content"],
            description=data["description"],
            status=data["status"]["status"],
            creator_id=data["creator"]["id"],
            creator_username=data["creator"]["username"],
            creator_email=data["creator"]["email"],
            assignees=data["assignees"],
            watchers=data["watchers"],
            priority=priority,
            due_date=data["due_date"],
            start_date=data["start_date"],
            points=data["points"],
            team_id=data["team_id"],
            project_id=data["project"]["id"],
        )





[docs]
@dataclass
class CUList(Component):
    """Component class for a list."""

    folder_id: float
    name: str
    content: Optional[str] = None
    due_date: Optional[int] = None
    due_date_time: Optional[bool] = None
    priority: Optional[int] = None
    assignee: Optional[int] = None
    status: Optional[str] = None


[docs]
    @classmethod
    def from_data(cls, data: dict) -> "CUList":
        return cls(
            folder_id=data["folder_id"],
            name=data["name"],
            content=data.get("content"),
            due_date=data.get("due_date"),
            due_date_time=data.get("due_date_time"),
            priority=data.get("priority"),
            assignee=data.get("assignee"),
            status=data.get("status"),
        )





[docs]
@dataclass
class Member(Component):
    """Component class for a member."""

    id: int
    username: str
    email: str
    initials: str


[docs]
    @classmethod
    def from_data(cls, data: Dict) -> "Member":
        return cls(
            id=data["user"]["id"],
            username=data["user"]["username"],
            email=data["user"]["email"],
            initials=data["user"]["initials"],
        )





[docs]
@dataclass
class Team(Component):
    """Component class for a team."""

    id: int
    name: str
    members: List[Member]


[docs]
    @classmethod
    def from_data(cls, data: Dict) -> "Team":
        members = [Member.from_data(member_data) for member_data in data["members"]]
        return cls(id=data["id"], name=data["name"], members=members)





[docs]
@dataclass
class Space(Component):
    """Component class for a space."""

    id: int
    name: str
    private: bool
    enabled_features: Dict[str, Any]


[docs]
    @classmethod
    def from_data(cls, data: Dict[str, Any]) -> "Space":
        space_data = data["spaces"][0]
        enabled_features = {
            feature: value
            for feature, value in space_data["features"].items()
            if value["enabled"]
        }
        return cls(
            id=space_data["id"],
            name=space_data["name"],
            private=space_data["private"],
            enabled_features=enabled_features,
        )





[docs]
def parse_dict_through_component(
    data: dict, component: Type[Component], fault_tolerant: bool = False
) -> Dict:
    """Parse a dictionary by creating
    a component and then turning it back into a dictionary.

    This helps with two things
    1. Extract and format data from a dictionary according to schema
    2. Provide a central place to do this in a fault-tolerant way

    """
    try:
        return asdict(component.from_data(data))
    except Exception as e:
        if fault_tolerant:
            warning_str = f"""Error encountered while trying to parse
{str(data)}: {str(e)}\n Falling back to returning input data."""
            warnings.warn(warning_str)
            return data
        else:
            raise e




[docs]
def extract_dict_elements_from_component_fields(
    data: dict, component: Type[Component]
) -> dict:
    """Extract elements from a dictionary.

    Args:
        data: The dictionary to extract elements from.
        component: The component to extract elements from.

    Returns:
        A dictionary containing the elements from the input dictionary that are also
        in the component.
    """
    output = {}
    for attribute in fields(component):
        if attribute.name in data:
            output[attribute.name] = data[attribute.name]
    return output




[docs]
def load_query(
    query: str, fault_tolerant: bool = False
) -> Tuple[Optional[Dict], Optional[str]]:
    """Parse a JSON string and return the parsed object.

    If parsing fails, returns an error message.

    :param query: The JSON string to parse.
    :return: A tuple containing the parsed object or None and an error message or None.

    Exceptions:
        json.JSONDecodeError: If the input is not a valid JSON string.
    """
    try:
        return json.loads(query), None
    except json.JSONDecodeError as e:
        if fault_tolerant:
            return (
                None,
                f"""Input must be a valid JSON. Got the following error: {str(e)}. 
"Please reformat and try again.""",
            )
        else:
            raise e




[docs]
def fetch_first_id(data: dict, key: str) -> Optional[int]:
    """Fetch the first id from a dictionary."""
    if key in data and len(data[key]) > 0:
        if len(data[key]) > 1:
            warnings.warn(f"Found multiple {key}: {data[key]}. Defaulting to first.")
        return data[key][0]["id"]
    return None




[docs]
def fetch_data(url: str, access_token: str, query: Optional[dict] = None) -> dict:
    """Fetch data from a URL."""
    headers = {"Authorization": access_token}
    response = requests.get(url, headers=headers, params=query)
    response.raise_for_status()
    return response.json()




[docs]
def fetch_team_id(access_token: str) -> Optional[int]:
    """Fetch the team id."""
    url = f"{DEFAULT_URL}/team"
    data = fetch_data(url, access_token)
    return fetch_first_id(data, "teams")




[docs]
def fetch_space_id(team_id: int, access_token: str) -> Optional[int]:
    """Fetch the space id."""
    url = f"{DEFAULT_URL}/team/{team_id}/space"
    data = fetch_data(url, access_token, query={"archived": "false"})
    return fetch_first_id(data, "spaces")




[docs]
def fetch_folder_id(space_id: int, access_token: str) -> Optional[int]:
    """Fetch the folder id."""
    url = f"{DEFAULT_URL}/space/{space_id}/folder"
    data = fetch_data(url, access_token, query={"archived": "false"})
    return fetch_first_id(data, "folders")




[docs]
def fetch_list_id(space_id: int, folder_id: int, access_token: str) -> Optional[int]:
    """Fetch the list id."""
    if folder_id:
        url = f"{DEFAULT_URL}/folder/{folder_id}/list"
    else:
        url = f"{DEFAULT_URL}/space/{space_id}/list"

    data = fetch_data(url, access_token, query={"archived": "false"})

    # The structure to fetch list id differs based if its folderless
    if folder_id and "id" in data:
        return data["id"]
    else:
        return fetch_first_id(data, "lists")




[docs]
class ClickupAPIWrapper(BaseModel):
    """Wrapper for Clickup API."""

    access_token: Optional[str] = None
    team_id: Optional[str] = None
    space_id: Optional[str] = None
    folder_id: Optional[str] = None
    list_id: Optional[str] = None

    model_config = ConfigDict(
        extra="forbid",
    )


[docs]
    @classmethod
    def get_access_code_url(
        cls, oauth_client_id: str, redirect_uri: str = "https://google.com"
    ) -> str:
        """Get the URL to get an access code."""
        url = f"https://app.clickup.com/api?client_id={oauth_client_id}"
        return f"{url}&redirect_uri={redirect_uri}"



[docs]
    @classmethod
    def get_access_token(
        cls, oauth_client_id: str, oauth_client_secret: str, code: str
    ) -> Optional[str]:
        """Get the access token."""
        url = f"{DEFAULT_URL}/oauth/token"

        params = {
            "client_id": oauth_client_id,
            "client_secret": oauth_client_secret,
            "code": code,
        }

        response = requests.post(url, params=params)
        data = response.json()

        if "access_token" not in data:
            print(f"Error: {data}")  # noqa: T201
            if "ECODE" in data and data["ECODE"] == "OAUTH_014":
                url = ClickupAPIWrapper.get_access_code_url(oauth_client_id)
                print(  # noqa: T201
                    "You already used this code once. Generate a new one.",
                    f"Our best guess for the url to get a new code is:\n{url}",
                )
            return None

        return data["access_token"]


    @model_validator(mode="before")
    @classmethod
    def validate_environment(cls, values: Dict) -> Any:
        """Validate that api key and python package exists in environment."""
        values["access_token"] = get_from_dict_or_env(
            values, "access_token", "CLICKUP_ACCESS_TOKEN"
        )
        values["team_id"] = fetch_team_id(values["access_token"])
        values["space_id"] = fetch_space_id(values["team_id"], values["access_token"])
        values["folder_id"] = fetch_folder_id(
            values["space_id"], values["access_token"]
        )
        values["list_id"] = fetch_list_id(
            values["space_id"], values["folder_id"], values["access_token"]
        )

        return values


[docs]
    def attempt_parse_teams(self, input_dict: dict) -> Dict[str, List[dict]]:
        """Parse appropriate content from the list of teams."""
        parsed_teams: Dict[str, List[dict]] = {"teams": []}
        for team in input_dict["teams"]:
            try:
                team = parse_dict_through_component(team, Team, fault_tolerant=False)
                parsed_teams["teams"].append(team)
            except Exception as e:
                warnings.warn(f"Error parsing a team {e}")

        return parsed_teams



[docs]
    def get_headers(
        self,
    ) -> Mapping[str, Union[str, bytes]]:
        """Get the headers for the request."""
        if not isinstance(self.access_token, str):
            raise TypeError(f"Access Token: {self.access_token}, must be str.")

        headers = {
            "Authorization": str(self.access_token),
            "Content-Type": "application/json",
        }
        return headers



[docs]
    def get_default_params(self) -> Dict:
        return {"archived": "false"}



[docs]
    def get_authorized_teams(self) -> Dict[Any, Any]:
        """Get all teams for the user."""
        url = f"{DEFAULT_URL}/team"

        response = requests.get(url, headers=self.get_headers())

        data = response.json()
        parsed_teams = self.attempt_parse_teams(data)

        return parsed_teams



[docs]
    def get_folders(self) -> Dict:
        """
        Get all the folders for the team.
        """
        url = f"{DEFAULT_URL}/team/" + str(self.team_id) + "/space"
        params = self.get_default_params()
        response = requests.get(url, headers=self.get_headers(), params=params)
        return {"response": response}



[docs]
    def get_task(self, query: str, fault_tolerant: bool = True) -> Dict:
        """
        Retrieve a specific task.
        """

        params, error = load_query(query, fault_tolerant=True)
        if params is None:
            return {"Error": error}

        url = f"{DEFAULT_URL}/task/{params['task_id']}"
        params = {
            "custom_task_ids": "true",
            "team_id": self.team_id,
            "include_subtasks": "true",
        }
        response = requests.get(url, headers=self.get_headers(), params=params)
        data = response.json()
        parsed_task = parse_dict_through_component(
            data, Task, fault_tolerant=fault_tolerant
        )

        return parsed_task



[docs]
    def get_lists(self) -> Dict:
        """
        Get all available lists.
        """

        url = f"{DEFAULT_URL}/folder/{self.folder_id}/list"
        params = self.get_default_params()
        response = requests.get(url, headers=self.get_headers(), params=params)
        return {"response": response}



[docs]
    def query_tasks(self, query: str) -> Dict:
        """
        Query tasks that match certain fields
        """
        params, error = load_query(query, fault_tolerant=True)
        if params is None:
            return {"Error": error}

        url = f"{DEFAULT_URL}/list/{params['list_id']}/task"

        params = self.get_default_params()
        response = requests.get(url, headers=self.get_headers(), params=params)

        return {"response": response}



[docs]
    def get_spaces(self) -> Dict:
        """
        Get all spaces for the team.
        """
        url = f"{DEFAULT_URL}/team/{self.team_id}/space"
        response = requests.get(
            url, headers=self.get_headers(), params=self.get_default_params()
        )
        data = response.json()
        parsed_spaces = parse_dict_through_component(data, Space, fault_tolerant=True)
        return parsed_spaces



[docs]
    def get_task_attribute(self, query: str) -> Dict:
        """
        Update an attribute of a specified task.
        """

        task = self.get_task(query, fault_tolerant=True)
        params, error = load_query(query, fault_tolerant=True)
        if not isinstance(params, dict):
            return {"Error": error}

        if params["attribute_name"] not in task:
            return {
                "Error": f"""attribute_name = {params["attribute_name"]} was not 
found in task keys {task.keys()}. Please call again with one of the key names."""
            }

        return {params["attribute_name"]: task[params["attribute_name"]]}



[docs]
    def update_task(self, query: str) -> Dict:
        """
        Update an attribute of a specified task.
        """
        query_dict, error = load_query(query, fault_tolerant=True)
        if query_dict is None:
            return {"Error": error}

        url = f"{DEFAULT_URL}/task/{query_dict['task_id']}"
        params = {
            "custom_task_ids": "true",
            "team_id": self.team_id,
            "include_subtasks": "true",
        }
        headers = self.get_headers()
        payload = {query_dict["attribute_name"]: query_dict["value"]}

        response = requests.put(url, headers=headers, params=params, json=payload)

        return {"response": response}



[docs]
    def update_task_assignees(self, query: str) -> Dict:
        """
        Add or remove assignees of a specified task.
        """
        query_dict, error = load_query(query, fault_tolerant=True)
        if query_dict is None:
            return {"Error": error}

        for user in query_dict["users"]:
            if not isinstance(user, int):
                return {
                    "Error": f"""All users must be integers, not strings!
"Got user {user} if type {type(user)}"""
                }

        url = f"{DEFAULT_URL}/task/{query_dict['task_id']}"

        headers = self.get_headers()

        if query_dict["operation"] == "add":
            assigne_payload = {"add": query_dict["users"], "rem": []}
        elif query_dict["operation"] == "rem":
            assigne_payload = {"add": [], "rem": query_dict["users"]}
        else:
            raise ValueError(
                f"Invalid operation ({query_dict['operation']}). ",
                "Valid options ['add', 'rem'].",
            )

        params = {
            "custom_task_ids": "true",
            "team_id": self.team_id,
            "include_subtasks": "true",
        }

        payload = {"assignees": assigne_payload}
        response = requests.put(url, headers=headers, params=params, json=payload)
        return {"response": response}



[docs]
    def create_task(self, query: str) -> Dict:
        """
        Creates a new task.
        """
        query_dict, error = load_query(query, fault_tolerant=True)
        if query_dict is None:
            return {"Error": error}

        list_id = self.list_id
        url = f"{DEFAULT_URL}/list/{list_id}/task"
        params = {"custom_task_ids": "true", "team_id": self.team_id}

        payload = extract_dict_elements_from_component_fields(query_dict, Task)
        headers = self.get_headers()

        response = requests.post(url, json=payload, headers=headers, params=params)
        data: Dict = response.json()
        return parse_dict_through_component(data, Task, fault_tolerant=True)



[docs]
    def create_list(self, query: str) -> Dict:
        """
        Creates a new list.
        """
        query_dict, error = load_query(query, fault_tolerant=True)
        if query_dict is None:
            return {"Error": error}

        # Default to using folder as location if it exists.
        # If not, fall back to using the space.
        location = self.folder_id if self.folder_id else self.space_id
        url = f"{DEFAULT_URL}/folder/{location}/list"

        payload = extract_dict_elements_from_component_fields(query_dict, Task)
        headers = self.get_headers()

        response = requests.post(url, json=payload, headers=headers)
        data = response.json()
        parsed_list = parse_dict_through_component(data, CUList, fault_tolerant=True)
        # set list id to new list
        if "id" in parsed_list:
            self.list_id = parsed_list["id"]
        return parsed_list



[docs]
    def create_folder(self, query: str) -> Dict:
        """
        Creates a new folder.
        """

        query_dict, error = load_query(query, fault_tolerant=True)
        if query_dict is None:
            return {"Error": error}

        space_id = self.space_id
        url = f"{DEFAULT_URL}/space/{space_id}/folder"
        payload = {
            "name": query_dict["name"],
        }

        headers = self.get_headers()

        response = requests.post(url, json=payload, headers=headers)
        data = response.json()

        if "id" in data:
            self.list_id = data["id"]
        return data



[docs]
    def run(self, mode: str, query: str) -> str:
        """Run the API."""
        if mode == "get_task":
            output = self.get_task(query)
        elif mode == "get_task_attribute":
            output = self.get_task_attribute(query)
        elif mode == "get_teams":
            output = self.get_authorized_teams()
        elif mode == "create_task":
            output = self.create_task(query)
        elif mode == "create_list":
            output = self.create_list(query)
        elif mode == "create_folder":
            output = self.create_folder(query)
        elif mode == "get_lists":
            output = self.get_lists()
        elif mode == "get_folders":
            output = self.get_folders()
        elif mode == "get_spaces":
            output = self.get_spaces()
        elif mode == "update_task":
            output = self.update_task(query)
        elif mode == "update_task_assignees":
            output = self.update_task_assignees(query)
        else:
            output = {"ModeError": f"Got unexpected mode {mode}."}

        try:
            return json.dumps(output)
        except Exception:
            return str(output)