python-slack-sdk/slack_sdk/models/messages/chunk.py at 2fc6abb9b191a38dce7d342e8dc1adbe732e9828 · 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
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 == 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 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