Automations

• 1: Automation
• 2: DoNothing
• 3: MetricChangeFilter
• 4: MetricThresholdFilter
• 5: NewAutomation
• 6: OnAddArtifactAlias
• 7: OnCreateArtifact
• 8: OnLinkArtifact
• 9: OnRunMetric
• 10: SendNotification
• 11: SendWebhook

import wandb
from wandb.automations import OnRunMetric, RunEvent, SendNotification

api = wandb.Api()

project = api.project("<my-project>", entity="<my-team>")

# Use the first Slack integration for the team
slack_hook = next(api.slack_integrations(entity="<my-team>"))

# Create a trigger event
event = OnRunMetric(
     scope=project,
     filter=RunEvent.metric("custom-metric") > 10,
)

# Create an action that responds to the event
action = SendNotification.from_integration(slack_hook)

# Create the automation
automation = api.create_automation(
     event >> action,
     name="my-automation",
     description="Send a Slack message whenever 'custom-metric' exceeds 10.",
)

1 - Automation

class Automation

A local instance of a saved W&B automation.

method Automation.__init__

__init__(
    typename__: 'Literal['Trigger']' = 'Trigger',
    id: 'str',
    created_at: 'datetime',
    updated_at: 'datetime | None' = None,
    name: 'str',
    description: 'str | None',
    enabled: 'bool',
    scope: '_ArtifactSequenceScope | _ArtifactPortfolioScope | ProjectScope',
    event: 'SavedEvent',
    action: 'SavedLaunchJobAction | SavedNotificationAction | SavedWebhookAction | SavedNoOpAction'
) → None

Args:

• typename__ (Literal[‘Trigger’]):
• id (str):
• created_at (datetime): The date and time when this automation was created.
• updated_at (Optional[datetime]): The date and time when this automation was last updated, if applicable.
• name (str): The name of this automation.
• description (Optional[str]): An optional description of this automation.
• enabled (bool): Whether this automation is enabled. Only enabled automations will trigger.
• scope (Union[_ArtifactSequenceScope, _ArtifactPortfolioScope, ProjectScope]): The scope in which the triggering event must occur.
• event (SavedEvent): The event that will trigger this automation.
• action (Union[SavedLaunchJobAction, SavedNotificationAction, SavedWebhookAction, SavedNoOpAction]): The action that will execute when this automation is triggered.

Returns: An Automation object.

2 - DoNothing

class DoNothing

Defines an automation action that intentionally does nothing.

method DoNothing.__init__

__init__(
    no_op: 'bool' = True,
    action_type: 'Literal[NO_OP]' = NO_OP
) → None

Args:

• no_op (bool): Placeholder field which exists only to satisfy backend schema requirements. There should never be a need to set this field explicitly, as its value is ignored.
• action_type (Literal[NO_OP]):

Returns: An DoNothing object.

3 - MetricChangeFilter

class MetricChangeFilter

Defines a filter that compares a change in a run metric against a user-defined threshold.

The change is calculated over “tumbling” windows, i.e. the difference between the current window and the non-overlapping prior window.

method MetricChangeFilter.__init__

__init__(
    name: 'str',
    agg: 'Agg | None' = None,
    window: 'int' = 1,
    cmp: 'None' = None,
    threshold: 'Annotated | Annotated',
    prior_window: 'int' = None,
    change_type: 'ChangeType',
    change_dir: 'ChangeDir'
) → None

Args:

• name (str):
• agg (Optional[Agg]):
• window (int):
• cmp (None): Ignored.
• threshold (Union[Annotated, Annotated]):
• prior_window (int): Size of the prior window over which the metric is aggregated (ignored if agg is None). If omitted, defaults to the size of the current window.
• change_type (ChangeType):
• change_dir (ChangeDir):

Returns: An MetricChangeFilter object.

4 - MetricThresholdFilter

class MetricThresholdFilter

Defines a filter that compares a run metric against a user-defined threshold value.

method MetricThresholdFilter.__init__

__init__(
    name: 'str',
    agg: 'Agg | None' = None,
    window: 'int' = 1,
    cmp: 'Literal['$gte', '$gt', '$lt', '$lte']',
    threshold: 'Annotated | Annotated'
) → None

Args:

• name (str):
• agg (Optional[Agg]):
• window (int):
• cmp (Literal[’$gte’, ‘$gt’, ‘$lt’, ‘$lte’]): Comparison operator used to compare the metric value (left) vs. the threshold value (right).
• threshold (Union[Annotated, Annotated]):

Returns: An MetricThresholdFilter object.

5 - NewAutomation

class NewAutomation

A new automation to be created.

method NewAutomation.__init__

__init__(
    name: 'str | None' = None,
    description: 'str | None' = None,
    enabled: 'bool | None' = None,
    event: 'Annotated | None' = None,
    action: 'Annotated | None' = None
) → None

Args:

• name (Optional[str]): The name of this automation.
• description (Optional[str]): An optional description of this automation.
• enabled (Optional[bool]): Whether this automation is enabled. Only enabled automations will trigger.
• event (Optional[Annotated]): The event that will trigger this automation.
• action (Optional[Annotated]): The action that will execute when this automation is triggered.

Returns: An NewAutomation object.

property NewAutomation.scope

The scope in which the triggering event must occur.

Returns:

• Optional[AutomationScope]: The scope property value.

6 - OnAddArtifactAlias

class OnAddArtifactAlias

A new alias is assigned to an artifact.

method OnAddArtifactAlias.__init__

__init__(
    event_type: 'Literal[ADD_ARTIFACT_ALIAS]' = ADD_ARTIFACT_ALIAS,
    scope: '_ArtifactSequenceScope | _ArtifactPortfolioScope | ProjectScope',
    filter: 'And | Or | Nor | Not | Lt | Gt | Lte | Gte | Eq | Ne | In | NotIn | Exists | Regex | Contains | dict[str, Any] | FilterExpr' = And([])
) → None

Args:

• event_type (Literal[ADD_ARTIFACT_ALIAS]):
• scope (Union[_ArtifactSequenceScope, _ArtifactPortfolioScope, ProjectScope]): The scope of the event.
• filter (Union[And, Or, Nor, Not, Lt, Gt, Lte, Gte, Eq, Ne, In, NotIn, Exists, Regex, Contains, Dict[str, Any], FilterExpr]): Additional condition(s), if any, that must be met for this event to trigger an automation.

Returns: An OnAddArtifactAlias object.

7 - OnCreateArtifact

class OnCreateArtifact

A new artifact is created.

method OnCreateArtifact.__init__

__init__(
    event_type: 'Literal[CREATE_ARTIFACT]' = CREATE_ARTIFACT,
    scope: '_ArtifactSequenceScope | _ArtifactPortfolioScope',
    filter: 'And | Or | Nor | Not | Lt | Gt | Lte | Gte | Eq | Ne | In | NotIn | Exists | Regex | Contains | dict[str, Any] | FilterExpr' = And([])
) → None

Args:

• event_type (Literal[CREATE_ARTIFACT]):
• scope (Union[_ArtifactSequenceScope, _ArtifactPortfolioScope]): The scope of the event: only artifact collections are valid scopes for this event.
• filter (Union[And, Or, Nor, Not, Lt, Gt, Lte, Gte, Eq, Ne, In, NotIn, Exists, Regex, Contains, Dict[str, Any], FilterExpr]): Additional condition(s), if any, that must be met for this event to trigger an automation.

Returns: An OnCreateArtifact object.

8 - OnLinkArtifact

class OnLinkArtifact

A new artifact is linked to a collection.

method OnLinkArtifact.__init__

__init__(
    event_type: 'Literal[LINK_ARTIFACT]' = LINK_ARTIFACT,
    scope: '_ArtifactSequenceScope | _ArtifactPortfolioScope | ProjectScope',
    filter: 'And | Or | Nor | Not | Lt | Gt | Lte | Gte | Eq | Ne | In | NotIn | Exists | Regex | Contains | dict[str, Any] | FilterExpr' = And([])
) → None

Args:

• event_type (Literal[LINK_ARTIFACT]):
• scope (Union[_ArtifactSequenceScope, _ArtifactPortfolioScope, ProjectScope]): The scope of the event.
• filter (Union[And, Or, Nor, Not, Lt, Gt, Lte, Gte, Eq, Ne, In, NotIn, Exists, Regex, Contains, Dict[str, Any], FilterExpr]): Additional condition(s), if any, that must be met for this event to trigger an automation.

Returns: An OnLinkArtifact object.

9 - OnRunMetric

class OnRunMetric

A run metric satisfies a user-defined condition.

method OnRunMetric.__init__

__init__(
    event_type: 'Literal[RUN_METRIC_THRESHOLD, RUN_METRIC_CHANGE]',
    scope: 'ProjectScope',
    filter: 'RunMetricFilter'
) → None

Args:

• event_type (Literal[RUN_METRIC_THRESHOLD, RUN_METRIC_CHANGE]):
• scope (ProjectScope): The scope of the event: only projects are valid scopes for this event.
• filter (RunMetricFilter): Run and/or metric condition(s) that must be satisfied for this event to trigger an automation.

Returns: An OnRunMetric object.

10 - SendNotification

class SendNotification

Defines an automation action that sends a (Slack) notification.

method SendNotification.__init__

__init__(
    integration_id: 'str',
    title: 'str' = '',
    message: 'str' = '',
    severity: 'AlertSeverity' = <AlertSeverity.INFO: 'INFO'>,
    action_type: 'Literal[NOTIFICATION]' = NOTIFICATION
) → None

Args:

• integration_id (str): The ID of the Slack integration that will be used to send the notification.
• title (str): The title of the sent notification.
• message (str): The message body of the sent notification.
• severity (AlertSeverity): The severity (INFO, WARN, ERROR) of the sent notification.
• action_type (Literal[NOTIFICATION]):

Returns: An SendNotification object.

classmethod SendNotification.from_integration

from_integration(
    integration: 'SlackIntegration',
    title: 'str' = '',
    text: 'str' = '',
    level: 'AlertSeverity' = <AlertSeverity.INFO: 'INFO'>
) → Self

Define a notification action that sends to the given (Slack) integration.

11 - SendWebhook

class SendWebhook

Defines an automation action that sends a webhook request.

method SendWebhook.__init__

__init__(
    integration_id: 'str',
    request_payload: 'Annotated | None' = None,
    action_type: 'Literal[GENERIC_WEBHOOK]' = GENERIC_WEBHOOK
) → None

Args:

• integration_id (str): The ID of the webhook integration that will be used to send the request.
• request_payload (Optional[Annotated]): The payload, possibly with template variables, to send in the webhook request.
• action_type (Literal[GENERIC_WEBHOOK]):

Returns: An SendWebhook object.

classmethod SendWebhook.from_integration

from_integration(
    integration: 'WebhookIntegration',
    payload: 'Optional[SerializedToJson[dict[str, Any]]]' = None
) → Self

Define a webhook action that sends to the given (webhook) integration.