{
  "nbformat": 4,
  "nbformat_minor": 5,
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "name": "python",
      "version": "3.10.0"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# W&B Automations tutorial (API)\n",
        "\n",
        "This notebook walks through managing W&B Automations with the Python API: list, get, create, update, and delete. It creates two example automations:\n",
        "\n",
        "1. **Project automation**: Run failure alert (Slack notification when a run fails).\n",
        "2. **Registry automation**: Alias added to webhook (when `production` alias is added to an artifact in a collection, trigger a webhook).\n",
        "\n",
        "**Prerequisites:** A W&B project, a Slack integration and a webhook configured in Team Settings, and (for the registry example) an artifact collection. Replace `my-team`, `my-project`, and `my-model` with your entity, project, and collection names."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## List automations"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import wandb\n",
        "\n",
        "api = wandb.Api()\n",
        "\n",
        "for automation in api.automations(entity=\"my-team\"):\n",
        "    print(automation.name, \"|\", getattr(automation, \"scope\", \"\"), \"|\", getattr(automation, \"enabled\", True))"
      ],
      "outputs": [],
      "execution_count": null
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Get one automation"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "automation = api.automation(name=\"run-failure-alert\", entity=\"my-team\")\n",
        "print(automation.name, automation.description, automation.enabled)"
      ],
      "outputs": [],
      "execution_count": null
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Create: Project automation (run failure to Slack)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from wandb.automations import OnRunState, RunEvent, SendNotification\n",
        "\n",
        "project = api.project(\"my-project\", entity=\"my-team\")\n",
        "slack_integration = next(api.slack_integrations(entity=\"my-team\"))\n",
        "\n",
        "event = OnRunState(\n",
        "    scope=project,\n",
        "    filter=RunEvent.state.in_([\"failed\"]),\n",
        ")\n",
        "action = SendNotification.from_integration(slack_integration)\n",
        "\n",
        "automation = api.create_automation(\n",
        "    event >> action,\n",
        "    name=\"run-failure-alert\",\n",
        "    description=\"Notify the team when a run fails.\",\n",
        ")\n",
        "print(\"Created:\", automation.name)"
      ],
      "outputs": [],
      "execution_count": null
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Create: Registry automation (alias added to webhook)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from wandb.automations import OnAddArtifactAlias, ArtifactEvent, SendWebhook\n",
        "\n",
        "collection = api.artifact_collection(name=\"my-model\", type_name=\"model\")\n",
        "webhook_integration = next(api.webhook_integrations(entity=\"my-team\"))\n",
        "\n",
        "event = OnAddArtifactAlias(\n",
        "    scope=collection,\n",
        "    filter=ArtifactEvent.alias.eq(\"production\"),\n",
        ")\n",
        "action = SendWebhook.from_integration(\n",
        "    webhook_integration,\n",
        "    payload={\n",
        "        \"event\": \"${event_type}\",\n",
        "        \"model\": \"${artifact_collection_name}\",\n",
        "        \"version\": \"${artifact_version_string}\",\n",
        "    },\n",
        ")\n",
        "\n",
        "automation = api.create_automation(\n",
        "    event >> action,\n",
        "    name=\"production-alias-webhook\",\n",
        "    description=\"Trigger webhook when production alias is added.\",\n",
        ")\n",
        "print(\"Created:\", automation.name)"
      ],
      "outputs": [],
      "execution_count": null
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Update an automation"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "automation = api.automation(name=\"run-failure-alert\", entity=\"my-team\")\n",
        "updated = api.update_automation(\n",
        "    automation,\n",
        "    enabled=False,\n",
        "    description=\"Temporarily disabled.\",\n",
        ")\n",
        "print(\"Updated:\", updated.name, updated.enabled, updated.description)"
      ],
      "outputs": [],
      "execution_count": null
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Delete an automation"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "# Uncomment to delete. Replace with your automation name.\n",
        "# api.delete_automation(api.automation(name=\"run-failure-alert\", entity=\"my-team\"))\n",
        "# Or by ID: api.delete_automation(automation.id)"
      ],
      "outputs": [],
      "execution_count": null
    }
  ]
}