Trigger dbt Cloud Job.ipynb

trigger-dbt-cloud-job.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Trigger dbt Cloud Job.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "authorship_tag": "ABX9TyNmO5kBL5LlsS0pRd/gSILk",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/gist/ernestoongaro/657689941acbf46d71f77b88627317bb/trigger-dbt-cloud-job.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yIqMkWPFKUeJ"
      },
      "source": [
        "**Triggering a dbt Cloud Job in your automated workflow with python (live example)**\n",
        "\n",
        "Script from [this dbt Labs discourse post](https://discourse.getdbt.com/t/triggering-a-dbt-cloud-job-in-your-automated-workflow-with-python/2573) by @[boxysean](https://discourse.getdbt.com/u/boxysean)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "cVK0zlgRJnDl"
      },
      "source": [
        "import enum\n",
        "import os\n",
        "import time\n",
        "\n",
        "import requests\n",
        "from getpass import getpass"
      ],
      "execution_count": 10,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "TUB34v-PJzSI",
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "outputId": "f7a29fab-701d-4fe6-b045-8c388173918e"
      },
      "source": [
        "ACCOUNT_ID = 4238\n",
        "JOB_ID = 12389\n",
        "\n",
        "\n",
        "\n",
        "API_KEY = getpass('Enter your Service Token key (found in Account Settings / Service Token): ')"
      ],
      "execution_count": 11,
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Enter your Service Token key (found in Account Settings / Service Token): ··········\n"
          ]
        }
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "c3S4C4zLKCQm",
        "outputId": "6c046176-6741-4fc2-9543-d17b28aa48f4"
      },
      "source": [
        "# These are documented on the dbt Cloud API docs\n",
        "class DbtJobRunStatus(enum.IntEnum):\n",
        "    QUEUED = 1\n",
        "    STARTING = 2\n",
        "    RUNNING = 3\n",
        "    SUCCESS = 10\n",
        "    ERROR = 20\n",
        "    CANCELLED = 30\n",
        "\n",
        "\n",
        "def _trigger_job() -> int:\n",
        "    res = requests.post(\n",
        "        url=f\"https://cloud.getdbt.com/api/v2/accounts/{ACCOUNT_ID}/jobs/{JOB_ID}/run/\",\n",
        "        headers={'Authorization': f\"Token {API_KEY}\"},\n",
        "        json={\n",
        "            # Optionally pass a description that can be viewed within the dbt Cloud API.\n",
        "            # See the API docs for additional parameters that can be passed in,\n",
        "            # including `schema_override` \n",
        "            'cause': f\"Triggered by my workflow!\",\n",
        "        }\n",
        "    )\n",
        "\n",
        "    try:\n",
        "        res.raise_for_status()\n",
        "    except:\n",
        "        print(f\"API token (last four): ...{API_KEY[-4:]}\")\n",
        "        raise\n",
        "\n",
        "    response_payload = res.json()\n",
        "    return response_payload['data']['id']\n",
        "\n",
        "\n",
        "def _get_job_run_status(job_run_id):\n",
        "    res = requests.get(\n",
        "        url=f\"https://cloud.getdbt.com/api/v2/accounts/{ACCOUNT_ID}/runs/{job_run_id}/\",\n",
        "        headers={'Authorization': f\"Token {API_KEY}\"},\n",
        "    )\n",
        "\n",
        "    res.raise_for_status()\n",
        "    response_payload = res.json()\n",
        "    return response_payload['data']['status']\n",
        "\n",
        "\n",
        "def run():\n",
        "    job_run_id = _trigger_job()\n",
        "\n",
        "    print(f\"job_run_id = {job_run_id}\")\n",
        "\n",
        "    \n",
        "    while True:\n",
        "        time.sleep(5)\n",
        "\n",
        "        status = _get_job_run_status(job_run_id)\n",
        "\n",
        "        print(DbtJobRunStatus(status))\n",
        "\n",
        "        if status == DbtJobRunStatus.SUCCESS:\n",
        "            break\n",
        "        elif status == DbtJobRunStatus.ERROR or status == DbtJobRunStatus.CANCELLED:\n",
        "            raise Exception(\"Failure!\")\n",
        "\n",
        "\n",
        "if __name__ == '__main__':\n",
        "    run()\n",
        "\n"
      ],
      "execution_count": 12,
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "job_run_id = 27744246\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.QUEUED\n",
            "DbtJobRunStatus.STARTING\n",
            "DbtJobRunStatus.STARTING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.RUNNING\n",
            "DbtJobRunStatus.SUCCESS\n"
          ]
        }
      ]
    }
  ]
}