python-slack-sdk/slack_sdk/models/messages/chunk.py at 33f745a1b3d4b11583af444e332d4f3f789f1d84 · slackapi/python-slack-sdk · GitHub

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
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
import logging
from typing import Dict, Optional, Sequence, Set, Union

from slack_sdk.models import show_unknown_key_warning
from slack_sdk.models.basic_objects import JsonObject
from slack_sdk.models.blocks.block_elements import UrlSourceElement


class Chunk(JsonObject):
    """
    Chunk for streaming messages.

    https://docs.slack.dev/messaging/sending-and-scheduling-messages#text-streaming
    """

    attributes = {"type"}
    logger = logging.getLogger(__name__)

    def __init__(
        self,
        *,
        type: Optional[str] = None,
    ):
        self.type = type

    @classmethod
    def parse(cls, chunk: Union[Dict, "Chunk"]) -> Optional["Chunk"]:
        if chunk is None:
            return None
        elif isinstance(chunk, Chunk):
            return chunk
        else:
            if "type" in chunk:
                type = chunk["type"]
                if type == MarkdownTextChunk.type:
                    return MarkdownTextChunk(**chunk)
                elif type == PlanUpdateChunk.type:
                    return PlanUpdateChunk(**chunk)
                elif type == TaskUpdateChunk.type:
                    return TaskUpdateChunk(**chunk)
                else:
                    cls.logger.warning(f"Unknown chunk detected and skipped ({chunk})")
                    return None
            else:
                cls.logger.warning(f"Unknown chunk detected and skipped ({chunk})")
                return None


class MarkdownTextChunk(Chunk):
    type = "markdown_text"

    @property
    def attributes(self) -> Set[str]:  # type: ignore[override]
        return super().attributes.union({"text"})

    def __init__(
        self,
        *,
        text: str,
        **others: Dict,
    ):
        """Used for streaming text content with markdown formatting support.

        https://docs.slack.dev/messaging/sending-and-scheduling-messages#text-streaming
        """
        super().__init__(type=self.type)
        show_unknown_key_warning(self, others)

        self.text = text


class PlanUpdateChunk(Chunk):
    type = "plan_update"

    @property
    def attributes(self) -> Set[str]:  # type: ignore[override]
        return super().attributes.union({"title"})

    def __init__(
        self,
        *,
        title: str,
        **others: Dict,
    ):
        """An updated title of plans for task and tool calls.

        https://docs.slack.dev/messaging/sending-and-scheduling-messages#text-streaming
        """
        super().__init__(type=self.type)
        show_unknown_key_warning(self, others)

        self.title = title


class TaskUpdateChunk(Chunk):
    type = "task_update"

    @property
    def attributes(self) -> Set[str]:  # type: ignore[override]
        return super().attributes.union(
            {
                "id",
                "title",
                "status",
                "details",
                "output",
                "sources",
            }
        )

    def __init__(
        self,
        *,
        id: str,
        title: str,
        status: str,  # "pending", "in_progress", "complete", "error"
        details: Optional[str] = None,
        output: Optional[str] = None,
        sources: Optional[Sequence[Union[Dict, UrlSourceElement]]] = None,
        **others: Dict,
    ):
        """Used for displaying tool execution progress in a timeline-style UI.

        https://docs.slack.dev/messaging/sending-and-scheduling-messages#text-streaming
        """
        super().__init__(type=self.type)
        show_unknown_key_warning(self, others)

        self.id = id
        self.title = title
        self.status = status
        self.details = details
        self.output = output
        self.sources = sources