vendor_connectors.slack

vendor_connectors.slack

Slack Connector using jbcom ecosystem packages.

Submodules

• vendor_connectors.slack.tools

Package Contents

Classes

SlackConnector	Slack connector for messaging, directory, and channel management.

Functions

get_divider	Return a Slack divider block.
get_header_block	Return header and divider blocks for a section title.
get_field_context_message_blocks	Build header and context blocks for detailed field data.
get_key_value_blocks	Format a key/value pair into Slack section blocks.
get_rich_text_blocks	Build a rich text block for multiline messages.

Data

MAX_RETRY_TIMEOUT_SECONDS

API

vendor_connectors.slack.MAX_RETRY_TIMEOUT_SECONDS

30

exception vendor_connectors.slack.SlackAPIError(response)

Bases: RuntimeError

Slack API error wrapper.

Initialization

Initialize self. See help(type(self)) for accurate signature.

vendor_connectors.slack.get_divider() → dict[str, str]

Return a Slack divider block.

Returns: dict[str, str]: Slack block definition for a divider element.

vendor_connectors.slack.get_header_block(field_title: str) → list[dict[str, Any]]

Return header and divider blocks for a section title.

Args: field_title: Title text to render in the header block.

Returns: list[dict[str, Any]]: Header block followed by a divider.

vendor_connectors.slack.get_field_context_message_blocks(field_name: str, context_data: collections.abc.Mapping) → list[dict[str, Any]]

Build header and context blocks for detailed field data.

Args: field_name: Name rendered in the header section. context_data: Mapping of key/value pairs rendered inside context blocks.

Returns: list[dict[str, Any]]: Blocks describing the field data.

vendor_connectors.slack.get_key_value_blocks(k: str, v: Any) → list[dict[str, Any]]

Format a key/value pair into Slack section blocks.

Args: k: Human-readable field label. v: Value to render. Mappings are encoded to Slack-safe text.

Returns: list[dict[str, Any]]: Section block followed by a divider.

vendor_connectors.slack.get_rich_text_blocks(lines: list[str], bold: bool = False, italic: bool = False, strike: bool = False) → list[dict[str, Any]]

Build a rich text block for multiline messages.

Args: lines: Message lines inserted as separate rich-text elements. bold: Whether to render text in bold. italic: Whether to render text in italics. strike: Whether to strike through the text.

Returns: list[dict[str, Any]]: Rich-text block followed by a divider.

class vendor_connectors.slack.SlackConnector(token: str | None = None, bot_token: str | None = None, logger: lifecyclelogging.Logging | None = None, **kwargs)

Bases: vendor_connectors.base.VendorConnectorBase

Slack connector for messaging, directory, and channel management.

Initialization

Initialize the Slack connector.

Args: token: Slack user token with directory scopes. bot_token: Bot token used for posting messages. logger: Optional shared logger instance. **kwargs: Extra keyword arguments forwarded to VendorConnectorBase.

static _normalize_identifier_filter(identifiers: str | collections.abc.Sequence[str] | None) → set[str] | None

Normalize comma-separated or iterable identifiers into a set.

Args: identifiers: Identifiers passed as a string or iterable.

Returns: Optional[set[str]]: Unique identifier set, or None when not provided.

send_message(channel_name: str, text: str, blocks: list | None = None, lines: list[str] | None = None, bold: bool = False, italic: bool = False, strike: bool = False, thread_id: str | None = None, raise_on_api_error: bool = True)

Send a message to a Slack channel using the bot token.

Args: channel_name: Human-readable channel name (without #). text: Plain text fallback for the message body. blocks: Optional structured block payload to include. lines: Convenience helper to render rich-text lines. bold: Whether to bold the rendered lines. italic: Whether to italicize the rendered lines. strike: Whether to strike-through the rendered lines. thread_id: Optional thread timestamp to reply within a thread. raise_on_api_error: When True, raise SlackAPIError on API failures.

Returns: str | Any: Timestamp string for the posted message or the Slack API response.

Raises: RuntimeError: If the bot is not a member of the channel. SlackAPIError: When Slack returns an error and raise_on_api_error is True.

get_bot_channels() → dict[str, dict]

Return channels the bot account is a member of.

Returns: dict[str, dict]: Mapping of channel name to channel metadata.

Raises: SlackAPIError: If Slack returns an error.

list_users(include_locale: bool | None = None, limit: int | None = None, team_id: str | None = None, include_deleted: bool | None = None, include_bots: bool | None = None, include_app_users: bool | None = None, **kwargs) → dict[str, dict[str, Any]]

List Slack users with optional filtering flags.

Args: include_locale: When True, include the locale for each user. limit: Maximum number of users per API call. team_id: Optional team/workspace ID. include_deleted: Include deactivated accounts when True. include_bots: Include bot accounts when True. include_app_users: Include app users when True. **kwargs: Additional keyword arguments forwarded to users_list.

Returns: dict[str, dict[str, Any]]: Filtered mapping of user IDs to user profiles.

list_usergroups(include_disabled: bool | None = None, include_count: bool | None = None, include_users: bool | None = None, team_id: str | None = None, usergroup_ids: str | collections.abc.Sequence[str] | None = None, **kwargs) → dict[str, dict[str, Any]]

List Slack user groups with optional filtering.

Args: include_disabled: Include disabled user groups when True. include_count: Include member counts when True. include_users: Include member lists when True. team_id: Optional workspace/team identifier. usergroup_ids: Comma-separated string or iterable of user group IDs to return. **kwargs: Extra keyword arguments forwarded to usergroups_list.

Returns: dict[str, dict[str, Any]]: Mapping of user group IDs to metadata.

list_conversations(exclude_archived: bool | None = None, limit: int | None = None, team_id: str | None = None, types: str | collections.abc.Sequence[str] | None = None, get_members: bool | None = None, channels_only: bool | None = None, **kwargs) → dict[str, dict[str, Any]]

List Slack conversations with optional filtering.

Args: exclude_archived: Exclude archived conversations when True. limit: Maximum number of conversations to request. team_id: Optional workspace/team identifier. types: Slack channel type(s) (public_channel, private_channel, im, mpim). get_members: Include member lists when True. channels_only: Return only channel-type conversations when True. **kwargs: Extra keyword arguments forwarded to conversations_list.

Returns: dict[str, dict[str, Any]]: Mapping of conversation IDs to metadata.

_call_api(method: str, group_by: str | None = None, id_field_name: str = ‘id’, **kwargs) → Any

Call a Slack WebClient method with retry and grouping support.

Args: method: Slack WebClient method name to invoke. group_by: Optional response field containing a list to re-index by ID. id_field_name: Field used as the dictionary key when grouping results. **kwargs: Keyword arguments forwarded to the Slack API method.

Returns: Any: Raw Slack response or grouped mapping when group_by is provided.

Raises: AttributeError: If the requested method is not implemented by WebClient. SlackAPIError: When Slack returns an error other than rate limiting. TimeoutError: If rate-limited retries exceed MAX_RETRY_TIMEOUT_SECONDS.