vendor_fabric.slack

Slack connector built on extended-data primitives.

Submodules

• vendor_fabric.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

SlackApiError
WebClient
MAX_RETRY_TIMEOUT_SECONDS

API

exception vendor_fabric.slack.SlackFallbackError

Bases: Exception

Fallback exception used until slack-sdk is imported.

Initialization

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

add_note()

class args

with_traceback()

vendor_fabric.slack.SlackApiError: Any = None

vendor_fabric.slack.WebClient: Any = None

vendor_fabric.slack.MAX_RETRY_TIMEOUT_SECONDS = 30

exception vendor_fabric.slack.SlackAPIError(response: Any)

Bases: RuntimeError

Slack API error wrapper.

Initialization

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

add_note()

class args

with_traceback()

vendor_fabric.slack.get_divider() → extended_data.containers.ExtendedDict

Return a Slack divider block.

Returns: Extended Slack block definition for a divider element.

vendor_fabric.slack.get_header_block(field_title: str) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

Return header and divider blocks for a section title.

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

Returns: Extended Slack blocks containing a header followed by a divider.

vendor_fabric.slack.get_field_context_message_blocks(field_name: str, context_data: collections.abc.Mapping[str, Any]) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

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: Extended Slack blocks describing the field data.

vendor_fabric.slack.get_key_value_blocks(k: str, v: Any) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

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: Extended Slack section block followed by a divider.

vendor_fabric.slack.get_rich_text_blocks(lines: list[str], bold: bool = False, italic: bool = False, strike: bool = False) → extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict]

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: Extended rich-text block followed by a divider.

class vendor_fabric.slack.SlackConnector(token: str | None = None, bot_token: str | None = None, logger: extended_data.logging.Logging | None = None, **kwargs: Any)

Bases: vendor_fabric.base.ConnectorBase

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 ConnectorBase.

send_message(channel_name: str, text: str, blocks: list[Any] | extended_data.containers.ExtendedList[extended_data.containers.ExtendedDict] | 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) → extended_data.containers.ExtendedString | extended_data.containers.ExtendedDict

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: Extended timestamp string, or an extended error payload when raise_on_api_error=False.

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() → extended_data.containers.ExtendedDict

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: Any) → extended_data.containers.ExtendedDict

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: Any) → extended_data.containers.ExtendedDict

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: Any) → extended_data.containers.ExtendedDict

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.

BASE_URL: ClassVar[str] = <Multiline-String>

API_KEY_ENV: ClassVar[str] = <Multiline-String>

CONNECTOR_CATEGORY: ClassVar[str] = 'external'

CONNECTOR_CAPABILITIES: ClassVar[tuple[str, ...]] = ()

TIMEOUT: ClassVar[float] = 300.0

MIN_REQUEST_INTERVAL: ClassVar[float] = 0.0

MAX_RETRIES: ClassVar[int] = 5

property api_key: str

property client: httpx.Client

close() → None

request(method: str, endpoint: str, *, headers: dict[str, str] | None = None, **kwargs: Any) → httpx.Response

decode_response(response: httpx.Response, *, suffix: str | None = None, as_extended: bool = True) → Any

decode_response_file(response: httpx.Response, *, source: str | None = None, suffix: str | None = None, as_extended: bool = True, metadata: collections.abc.Mapping[str, Any] | None = None) → extended_data.io.DataFile

extend_result(value: Any) → Any

request_data(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any

request_data_file(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.io.DataFile

request_workflow(method: str, endpoint: str, *, headers: dict[str, str] | None = None, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow

get(endpoint: str, **kwargs: Any) → httpx.Response

get_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any

get_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow

post(endpoint: str, **kwargs: Any) → httpx.Response

post_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any

post_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow

put(endpoint: str, **kwargs: Any) → httpx.Response

put_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any

put_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow

delete(endpoint: str, **kwargs: Any) → httpx.Response

delete_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any

delete_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow

patch(endpoint: str, **kwargs: Any) → httpx.Response

patch_data(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → Any

patch_workflow(endpoint: str, *, suffix: str | None = None, as_extended: bool = True, **kwargs: Any) → extended_data.workflows.DataWorkflow

download(url: str, output_path: str) → int

vendor_capabilities: ClassVar[dict[str, vendor_fabric.capabilities.CapabilitySpec]] = None

vendor_capability_methods: ClassVar[dict[str, str]] = None

get_input(k: str, default: Any | None = None, required: bool = False, is_bool: bool = False, is_integer: bool = False, is_float: bool = False, is_path: bool = False, is_datetime: bool = False, as_extended: bool = False) → Any

decode_input(k: str, default: Any | None = None, required: bool = False, decode_from_json: bool = False, decode_from_yaml: bool = False, decode_from_base64: bool = False, allow_none: bool = True, as_extended: bool = False) → Any

freeze_inputs() → extended_data.containers.mappings.ExtendedDict

thaw_inputs() → extended_data.containers.mappings.ExtendedDict

snapshot_inputs(*, frozen: bool = False) → extended_data.containers.mappings.ExtendedDict

replace_inputs(new_inputs: collections.abc.Mapping[str, Any] | None, *, clear_frozen: bool = True) → extended_data.containers.mappings.ExtendedDict

merge_inputs(new_inputs: collections.abc.Mapping[str, Any] | None) → extended_data.containers.mappings.ExtendedDict

shift_inputs() → extended_data.containers.mappings.ExtendedDict