plugins.h_slack.SlackNotifier

Provides a Slack notifier for Hamilton execution. Must have slack_sdk installed to use it:

pip install sf-hamilton[slack] (use quotes if using zsh)

class hamilton.plugins.h_slack.SlackNotifier(api_key: str, channel: str, **kwargs)

This is a adapter that sends a message to a slack channel when a node is executed & fails.

Note: you need to have slack_sdk installed for this to work. If you don’t have it installed, you can install it with pip install slack_sdk (or pip install sf-hamilton[slack] – use quotes if you’re using zsh).

from hamilton.plugins import h_slack

dr = (
    driver.Builder()
    .with_config({})
    .with_modules(some_modules)
    .with_adapters(h_slack.SlackNotifier(api_key="YOUR_API_KEY", channel="YOUR_CHANNEL"))
    .build()
)
# and then when you call .execute() or .materialize() you'll get a message in your slack channel!

__init__(api_key: str, channel: str, **kwargs)

Constructor.

Parameters:

• api_key – API key to use for sending messages.

• channel – Channel to send messages to.

post_node_execute(*, run_id: str, node_: Node, kwargs: Dict[str, Any], success: bool, error: Exception | None, result: Any | None, task_id: str | None = None)

Wraps the after_execution method, providing a bridge to an external-facing API. Do not override this!

pre_node_execute(*, run_id: str, node_: Node, kwargs: Dict[str, Any], task_id: str | None = None)

Wraps the before_execution method, providing a bridge to an external-facing API. Do not override this!

run_after_node_execution(node_name: str, node_tags: Dict[str, Any], node_kwargs: Dict[str, Any], node_return_type: type, result: Any, error: Exception | None, success: bool, task_id: str | None, run_id: str, **future_kwargs: Any)

Sends a message to the slack channel after a node is executed.

run_before_node_execution(node_name: str, node_tags: Dict[str, Any], node_kwargs: Dict[str, Any], node_return_type: type, **future_kwargs: Any)

Placeholder required to subclass NodeExecutionMethod