Skip to content

Asana

AsanaReader #

Bases: BaseReader

Asana阅读器。从Asana工作区读取数据。

Parameters:

Name Type Description Default
asana_token str

Asana令牌。

 required
Source code in llama_index/readers/asana/base.py 
  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
 94
 95
 96
 97
 98
 99
100
101
class AsanaReader(BaseReader):
    """Asana阅读器。从Asana工作区读取数据。

    Args:
        asana_token (str): Asana令牌。"""

    def __init__(self, asana_token: str) -> None:
        """初始化Asana阅读器。"""
        self.client = asana.Client.access_token(asana_token)

    def load_data(
        self, workspace_id: Optional[str] = None, project_id: Optional[str] = None
    ) -> List[Document]:
        """从工作空间加载数据。

Args:
    workspace_id (可选[str], 可选): 工作空间ID。默认为None。
    project_id (可选[str], 可选): 项目ID。默认为None。

Returns:
    List[Document]: 文档列表。
"""
        if workspace_id is None and project_id is None:
            raise ValueError("Either workspace_id or project_id must be provided")

        if workspace_id is not None and project_id is not None:
            raise ValueError(
                "Only one of workspace_id or project_id should be provided"
            )

        results = []

        if workspace_id is not None:
            workspace_name = self.client.workspaces.find_by_id(workspace_id)["name"]
            projects = self.client.projects.find_all({"workspace": workspace_id})

        # Case: Only project_id is provided
        else:  # since we've handled the other cases, this means project_id is not None
            projects = [self.client.projects.find_by_id(project_id)]
            workspace_name = projects[0]["workspace"]["name"]

        for project in projects:
            tasks = self.client.tasks.find_all(
                {
                    "project": project["gid"],
                    "opt_fields": "name,notes,completed,completed_at,completed_by,assignee,followers,custom_fields",
                }
            )
            for task in tasks:
                stories = self.client.tasks.stories(task["gid"], opt_fields="type,text")
                comments = "\n".join(
                    [
                        story["text"]
                        for story in stories
                        if story.get("type") == "comment" and "text" in story
                    ]
                )

                task_metadata = {
                    "task_id": task.get("gid", ""),
                    "name": task.get("name", ""),
                    "assignee": (task.get("assignee") or {}).get("name", ""),
                    "completed_on": task.get("completed_at", ""),
                    "completed_by": (task.get("completed_by") or {}).get("name", ""),
                    "project_name": project.get("name", ""),
                    "custom_fields": [
                        i["display_value"]
                        for i in task.get("custom_fields")
                        if task.get("custom_fields") is not None
                    ],
                    "workspace_name": workspace_name,
                    "url": f"https://app.asana.com/0/{project['gid']}/{task['gid']}",
                }

                if task.get("followers") is not None:
                    task_metadata["followers"] = [
                        i.get("name") for i in task.get("followers") if "name" in i
                    ]
                else:
                    task_metadata["followers"] = []

                results.append(
                    Document(
                        text=task.get("name", "")
                        + " "
                        + task.get("notes", "")
                        + " "
                        + comments,
                        extra_info=task_metadata,
                    )
                )

        return results

load_data #

load_data(
    workspace_id: Optional[str] = None,
    project_id: Optional[str] = None,
) -> List[Document]

从工作空间加载数据。

Parameters:

Name Type Description Default
workspace_id (可选[str], 可选)

工作空间ID。默认为None。

 None
project_id (可选[str], 可选)

项目ID。默认为None。

 None

Returns:

Type Description
List[Document]

List[Document]: 文档列表。
Source code in llama_index/readers/asana/base.py 
 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
 94
 95
 96
 97
 98
 99
100
101
    def load_data(
        self, workspace_id: Optional[str] = None, project_id: Optional[str] = None
    ) -> List[Document]:
        """从工作空间加载数据。

Args:
    workspace_id (可选[str], 可选): 工作空间ID。默认为None。
    project_id (可选[str], 可选): 项目ID。默认为None。

Returns:
    List[Document]: 文档列表。
"""
        if workspace_id is None and project_id is None:
            raise ValueError("Either workspace_id or project_id must be provided")

        if workspace_id is not None and project_id is not None:
            raise ValueError(
                "Only one of workspace_id or project_id should be provided"
            )

        results = []

        if workspace_id is not None:
            workspace_name = self.client.workspaces.find_by_id(workspace_id)["name"]
            projects = self.client.projects.find_all({"workspace": workspace_id})

        # Case: Only project_id is provided
        else:  # since we've handled the other cases, this means project_id is not None
            projects = [self.client.projects.find_by_id(project_id)]
            workspace_name = projects[0]["workspace"]["name"]

        for project in projects:
            tasks = self.client.tasks.find_all(
                {
                    "project": project["gid"],
                    "opt_fields": "name,notes,completed,completed_at,completed_by,assignee,followers,custom_fields",
                }
            )
            for task in tasks:
                stories = self.client.tasks.stories(task["gid"], opt_fields="type,text")
                comments = "\n".join(
                    [
                        story["text"]
                        for story in stories
                        if story.get("type") == "comment" and "text" in story
                    ]
                )

                task_metadata = {
                    "task_id": task.get("gid", ""),
                    "name": task.get("name", ""),
                    "assignee": (task.get("assignee") or {}).get("name", ""),
                    "completed_on": task.get("completed_at", ""),
                    "completed_by": (task.get("completed_by") or {}).get("name", ""),
                    "project_name": project.get("name", ""),
                    "custom_fields": [
                        i["display_value"]
                        for i in task.get("custom_fields")
                        if task.get("custom_fields") is not None
                    ],
                    "workspace_name": workspace_name,
                    "url": f"https://app.asana.com/0/{project['gid']}/{task['gid']}",
                }

                if task.get("followers") is not None:
                    task_metadata["followers"] = [
                        i.get("name") for i in task.get("followers") if "name" in i
                    ]
                else:
                    task_metadata["followers"] = []

                results.append(
                    Document(
                        text=task.get("name", "")
                        + " "
                        + task.get("notes", "")
                        + " "
                        + comments,
                        extra_info=task_metadata,
                    )
                )

        return results