Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[ENH] Add WebhookMessage class #70

Merged
merged 2 commits into from
Feb 11, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "slackblocks"
version = "1.0.2"
version = "1.0.3"
description = "Python wrapper for the Slack Blocks API"
authors = [
"Nicholas Lambourne <[email protected]>",
Expand Down
2 changes: 1 addition & 1 deletion slackblocks/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@
WorkflowButton,
)
from .errors import InvalidUsageError
from .messages import Message, MessageResponse
from .messages import Message, MessageResponse, ResponseType, WebhookMessage
from .modals import Modal
from .objects import (
Confirm,
Expand Down
123 changes: 122 additions & 1 deletion slackblocks/messages.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,34 @@
See: <https://api.slack.com/messaging>
"""

from enum import Enum
from json import dumps
from typing import Any, Dict, List, Optional, Union

from slackblocks.utils import coerce_to_list

from .attachments import Attachment
from .blocks import Block
from .errors import InvalidUsageError


class ResponseType(Enum):
"""
Types of messages that can be sent via `WebhookMessage`.
"""

EPHEMERAL = "ephemeral"
IN_CHANNEL = "in_channel"

@staticmethod
def get_value(value: Union["ResponseType", str]) -> str:
if isinstance(value, ResponseType):
return value.value
if value not in [response_type.value for response_type in ResponseType]:
raise InvalidUsageError(
"ResponseType must be either `ephemeral` or `in_channel`"
)
return value


class BaseMessage:
Expand Down Expand Up @@ -75,7 +96,7 @@ class Message(BaseMessage):
the Slack message API.

Args:
channel:
channel: the Slack channel to send the message to, e.g. "#general".
text: markdown text to send in the message. If `blocks` are provided
then this is a fallback to display in notifications.
blocks: a list of [`Blocks`](/reference/blocks) to form the contents
Expand Down Expand Up @@ -150,3 +171,103 @@ def _resolve(self) -> Dict[str, Any]:
if self.ephemeral:
result["response_type"] = "ephemeral"
return result


class WebhookMessage:
"""
Messages sent via the Slack `WebhookClient` takes different arguments than
those sent via the regular `WebClient`.

See: <https://github.com/slackapi/python-slack-sdk/blob/7e71b73/slack_sdk/webhook/client.py#L28>

Args:
text: markdown text to send in the message. If `blocks` are provided
then this is a fallback to display in notifications.
attachments: a list of
[`Attachments`](/reference/attachments/#attachments.Attachment)
that form the secondary contents of the message (deprecated).
blocks: a list of [`Blocks`](/reference/blocks) to form the contents
of the message instead of the contents of `text`.
response_type: one of `ResponseType.EPHEMERAL` or `ResponseType.IN_CHANNEL`.
Ephemeral messages are shown only to the requesting user whereas
"in-channel" messages are shown to all channel participants.
replace_orginal: when `True`, the message triggering this response will be
replaced by this messaage. Mutually exclusive with `delete_original`.
delete_original: when `True`, the original message triggering this response
will be deleted, and any content of this message will be posted as a
new message. Mutually exclusive with `replace_orginal`.
unfurl_links: if `True`, links in the message will be automatically
unfurled.
unfurl_media: if `True`, media from links (e.g. images) will
automatically unfurl.
metadata: additional metadata to attach to the message.
headres: HTTP request headers to include with the message.

Throws:
InvalidUsageError: when any of the passed fields fail validation.
"""

def __init__(
self,
text: Optional[str] = None,
attachments: Optional[List[Attachment]] = None,
blocks: Optional[Union[List[Block], Block]] = None,
response_type: Union[ResponseType, str] = None,
replace_original: Optional[bool] = None,
delete_original: Optional[bool] = None,
unfurl_links: Optional[bool] = None,
unfurl_media: Optional[bool] = None,
metadata: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
) -> "WebhookMessage":
self.text = text
self.attachments = coerce_to_list(attachments, Attachment, allow_none=True)
self.blocks = coerce_to_list(blocks, Block, allow_none=True)
self.response_type = ResponseType.get_value(response_type)
self.replace_original = replace_original
self.delete_original = delete_original
self.unfurl_links = unfurl_links
self.unfurl_media = unfurl_media
self.metadata = metadata
self.headers = headers

def _resolve(self) -> None:
webhook_message = {}
if self.text is not None:
webhook_message["text"] = self.text
if self.attachments is not None:
webhook_message["attachments"] = [
attachment._resolve() for attachment in self.attachments
]
if self.blocks is not None:
webhook_message["blocks"] = [block._resolve() for block in self.blocks]
if self.response_type is not None:
webhook_message["response_type"] = self.response_type
if self.replace_original is not None:
webhook_message["replace_original"] = self.replace_original
if self.delete_original is not None:
webhook_message["delete_original"] = self.delete_original
if self.unfurl_links is not None:
webhook_message["unfurl_links"] = self.unfurl_links
if self.unfurl_media is not None:
webhook_message["unfurl_media"] = self.unfurl_media
if self.metadata is not None:
webhook_message["metadata"] = self.metadata
if self.headers is not None:
webhook_message["headers"] = self.headers
return webhook_message

def to_dict(self) -> Dict[str, Any]:
return self._resolve()

def json(self) -> str:
return dumps(self._resolve(), indent=4)

def __repr__(self) -> str:
return self.json()

def __getitem__(self, item):
return self._resolve()[item]

def keys(self) -> Dict[str, Any]:
return self._resolve().keys()
27 changes: 27 additions & 0 deletions test/samples/messages/webhook_message_basic.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
{
"blocks": [
{
"type": "section",
"block_id": "fake_block_id",
"text": {
"type": "mrkdwn",
"text": "You wouldn't do ol' Hook in now, would you, lad?"
}
},
{
"type": "section",
"block_id": "fake_block_id",
"text": {
"type": "mrkdwn",
"text": "Well, all right... if you... say you're a codfish."
}
}
],
"response_type": "ephemeral",
"replace_original": true,
"unfurl_links": false,
"unfurl_media": false,
"metadata": {
"sender": "Walt"
}
}
41 changes: 41 additions & 0 deletions test/samples/messages/webhook_message_delete.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
{
"attachments": [
{
"blocks": [
{
"type": "section",
"block_id": "fake_block_id",
"text": {
"type": "mrkdwn",
"text": "I'M A CODFISH!"
}
}
]
}
],
"blocks": [
{
"type": "section",
"block_id": "fake_block_id",
"text": {
"type": "mrkdwn",
"text": "I'm a codfish."
}
},
{
"type": "section",
"block_id": "fake_block_id",
"text": {
"type": "mrkdwn",
"text": "Louder!"
}
}
],
"response_type": "in_channel",
"delete_original": true,
"unfurl_links": true,
"unfurl_media": true,
"metadata": {
"sender": "Walt"
}
}
77 changes: 76 additions & 1 deletion test/unit/test_messages.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,13 @@
from slackblocks import Attachment, Color, Message, MessageResponse, SectionBlock
from slackblocks import (
Attachment,
Color,
Message,
MessageResponse,
ResponseType,
SectionBlock,
Text,
WebhookMessage,
)


def test_basic_message() -> None:
Expand Down Expand Up @@ -58,3 +67,69 @@ def test_to_dict() -> None:
"replace_original": False,
"response_type": "ephemeral",
}


def test_basic_webhook_message() -> None:
with open("test/samples/messages/webhook_message_basic.json", "r") as expected:
assert (
repr(
WebhookMessage(
blocks=[
SectionBlock(
Text("You wouldn't do ol' Hook in now, would you, lad?"),
block_id="fake_block_id",
),
SectionBlock(
Text("Well, all right... if you... say you're a codfish."),
block_id="fake_block_id",
),
],
response_type=ResponseType.EPHEMERAL,
replace_original=True,
unfurl_links=False,
unfurl_media=False,
metadata={
"sender": "Walt",
},
)
)
== expected.read()
)


def test_webhook_message_delete() -> None:
with open("test/samples/messages/webhook_message_delete.json", "r") as expected:
assert (
repr(
WebhookMessage(
attachments=[
Attachment(
blocks=[
SectionBlock(
Text("I'M A CODFISH!"),
block_id="fake_block_id",
)
]
)
],
blocks=[
SectionBlock(
Text("I'm a codfish."),
block_id="fake_block_id",
),
SectionBlock(
Text("Louder!"),
block_id="fake_block_id",
),
],
response_type="in_channel",
delete_original=True,
unfurl_links=True,
unfurl_media=True,
metadata={
"sender": "Walt",
},
)
)
== expected.read()
)