Skip to content

Instantly share code, notes, and snippets.

@ernestoongaro

ernestoongaro/trigger-dbt-cloud-job.ipynb

Last active February 28, 2024 12:19
Show Gist options
  • Star 2 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
  • Save ernestoongaro/657689941acbf46d71f77b88627317bb to your computer and use it in GitHub Desktop.
Save ernestoongaro/657689941acbf46d71f77b88627317bb to your computer and use it in GitHub Desktop.
Download ZIP
Trigger dbt Cloud Job.ipynb
Display the source blob
Display the rendered blob
Raw
trigger-dbt-cloud-job.ipynb
{
"nbformat": 4,
"nbformat_minor": 0,
"metadata": {
"colab": {
"name": "Trigger dbt Cloud Job.ipynb",
"provenance": [],
"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": 1,
"outputs": []
},
{
"cell_type": "code",
"metadata": {
"id": "TUB34v-PJzSI",
"colab": {
"base_uri": "https://localhost:8080/"
},
"outputId": "71783103-37b5-40c6-f5c5-36f1e1e4d623"
},
"source": [
"ACCOUNT_ID = 131\n",
"JOB_ID = 10953\n",
"\n",
"\n",
"\n",
"API_KEY = getpass('Enter your Service Token key (found in Account Settings / Service Token): ')"
],
"execution_count": 2,
"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/",
"height": 356
},
"id": "c3S4C4zLKCQm",
"outputId": "8d58be7b-07fe-4993-be43-37a63c6c6135"
},
"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://emea.dbt.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://emea.dbt.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": 3,
"outputs": [
{
"output_type": "stream",
"name": "stdout",
"text": [
"job_run_id = 3749277\n"
]
},
{
"output_type": "error",
"ename": "HTTPError",
"evalue": "401 Client Error: Unauthorized for url: https://cloud.getdbt.com/api/v2/accounts/131/runs/3749277/",
"traceback": [
"\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
"\u001b[0;31mHTTPError\u001b[0m Traceback (most recent call last)",
"\u001b[0;32m<ipython-input-3-4ec271e2acc6>\u001b[0m in \u001b[0;36m<cell line: 63>\u001b[0;34m()\u001b[0m\n\u001b[1;32m 62\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m 63\u001b[0m \u001b[0;32mif\u001b[0m \u001b[0m__name__\u001b[0m \u001b[0;34m==\u001b[0m \u001b[0;34m'__main__'\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 64\u001b[0;31m \u001b[0mrun\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m 65\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n",
"\u001b[0;32m<ipython-input-3-4ec271e2acc6>\u001b[0m in \u001b[0;36mrun\u001b[0;34m()\u001b[0m\n\u001b[1;32m 51\u001b[0m \u001b[0mtime\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0msleep\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;36m5\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m 52\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 53\u001b[0;31m \u001b[0mstatus\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0m_get_job_run_status\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mjob_run_id\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m 54\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m 55\u001b[0m \u001b[0mprint\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mDbtJobRunStatus\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mstatus\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
"\u001b[0;32m<ipython-input-3-4ec271e2acc6>\u001b[0m in \u001b[0;36m_get_job_run_status\u001b[0;34m(job_run_id)\u001b[0m\n\u001b[1;32m 37\u001b[0m )\n\u001b[1;32m 38\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 39\u001b[0;31m \u001b[0mres\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mraise_for_status\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m 40\u001b[0m \u001b[0mresponse_payload\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mres\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mjson\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m 41\u001b[0m \u001b[0;32mreturn\u001b[0m \u001b[0mresponse_payload\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m'data'\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m[\u001b[0m\u001b[0;34m'status'\u001b[0m\u001b[0;34m]\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
"\u001b[0;32m/usr/local/lib/python3.10/dist-packages/requests/models.py\u001b[0m in \u001b[0;36mraise_for_status\u001b[0;34m(self)\u001b[0m\n\u001b[1;32m 1019\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m 1020\u001b[0m \u001b[0;32mif\u001b[0m \u001b[0mhttp_error_msg\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m-> 1021\u001b[0;31m \u001b[0;32mraise\u001b[0m \u001b[0mHTTPError\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mhttp_error_msg\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mresponse\u001b[0m\u001b[0;34m=\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m 1022\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m 1023\u001b[0m \u001b[0;32mdef\u001b[0m \u001b[0mclose\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0mself\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
"\u001b[0;31mHTTPError\u001b[0m: 401 Client Error: Unauthorized for url: https://cloud.getdbt.com/api/v2/accounts/131/runs/3749277/"
]
}
]
}
]
}
@ernestoongaro
Copy link
Author

Updated to reflect that we can use Service Token instead of Profile token. Thx @boxysean

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment