usage/manage_pipeline.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "f44f86fc-03d2-4eae-971d-73f6767c97a8",
   "metadata": {
    "collapsed": true,
    "jupyter": {
     "outputs_hidden": true
    }
   },
   "source": [
    "# Manage GlassFlow pipelines\n",
    "\n",
    "In this notebook, we demonstrate how to manage pipelines using the SDK.\n",
    "\n",
    "We begin by creating a client with our Personal Access Token"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "c1e018bc-dae2-437a-bdba-653f77dbda59",
   "metadata": {},
   "outputs": [],
   "source": [
    "from glassflow import GlassFlowClient\n",
    "\n",
    "# Please edit this variable with your own personal access token from https://app.glassflow.dev/profile\n",
    "personal_access_token = \"< your-personal-access-token >\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "2b172e74-636a-4dc9-a034-bd1c0834bcaf",
   "metadata": {},
   "outputs": [],
   "source": [
    "client = GlassFlowClient(\n",
    "    personal_access_token=personal_access_token\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4e1c0276-c446-42d0-8de7-056344a187c8",
   "metadata": {},
   "source": [
    "## Create pipeline\n",
    "\n",
    "If you have already a pipeline created you can skip this section and jump to [Update pipeline](#update-pipeline) section."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "a1238d51-39aa-4967-9139-dcaa1f060a4b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing transformation function ...\n",
      "Done!\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "echo \"Writing transformation function ...\"\n",
    "cat <<EOT > transformation.py \n",
    "def handler(data, log):\n",
    "    log.info(data)\n",
    "    return data\n",
    "EOT\n",
    "echo \"Done!\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "3e0f754f-9862-49eb-810b-bdce21187785",
   "metadata": {},
   "outputs": [],
   "source": [
    "space = client.create_space(name=\"tutorials\")\n",
    "pipeline = client.create_pipeline(\n",
    "    name=\"example pipeline\",\n",
    "    space_id=space.id,\n",
    "    transformation_file=\"transformation.py\",\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "90c8ef94-8d4c-4442-adb5-a0bf398327d2",
   "metadata": {},
   "source": [
    "## Update pipeline\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "9fb8a5f7-c194-47fe-8ed3-a1e072fa2bcb",
   "metadata": {},
   "outputs": [],
   "source": [
    "PIPELINE_ID = pipeline.id\n",
    "pipeline_to_update = client.get_pipeline(pipeline_id=PIPELINE_ID)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c34eb5fe-2228-484b-8282-5a420b4b623a",
   "metadata": {},
   "source": [
    "Let's update the transformation code and add an environment variable to the pipeline."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "ddff068f-9499-4651-adc6-64cd8b74f372",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing transformation function ...\n",
      "Done!\n"
     ]
    }
   ],
   "source": [
    "%%bash\n",
    "echo \"Writing transformation function ...\"\n",
    "cat <<EOT > transformation_with_env.py \n",
    "import os\n",
    "\n",
    "\n",
    "def handler(data, log):\n",
    "    env = os.getenv(\"MY_VAR\")\n",
    "    log.info(\"My env variable: %s\", env)\n",
    "    return data\n",
    "EOT\n",
    "echo \"Done!\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "7332b5b2-d632-4d24-b93b-638bd3caeb10",
   "metadata": {},
   "outputs": [],
   "source": [
    "updated_pipeline = pipeline_to_update.update(\n",
    "    transformation_file=\"transformation_with_env.py\",\n",
    "    env_vars=[\n",
    "        {\n",
    "            \"name\": \"MY_VAR\",\n",
    "            \"value\": \"my env var value\"\n",
    "        }\n",
    "    ]\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "46c90d22-d939-4db0-8ad4-16005c3cce08",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[{'name': 'MY_VAR', 'value': 'my env var value'}]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "updated_pipeline.env_vars"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "b2698188-314c-4402-b21b-737c77596ab1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "```python\n",
       "import os\n",
       "\n",
       "\n",
       "def handler(data, log):\n",
       "    env = os.getenv(\"MY_VAR\")\n",
       "    log.info(\"My env variable: %s\", env)\n",
       "    return data\n",
       "\n",
       "```"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from IPython.display import Markdown as md\n",
    "\n",
    "md(f\"```python\\n{updated_pipeline.transformation_code}\\n```\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "86da7c19-2c17-483f-b1ca-361f8462c85b",
   "metadata": {},
   "source": [
    "### Pause a pipeline\n",
    "\n",
    "If you want to pause the pipeline so it does not process any new event, we can do that as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "0b8b5cc2-82e1-49e0-8220-ac3dec3e7bef",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'paused'"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "updated_pipeline = pipeline_to_update.update(\n",
    "    state=\"paused\"\n",
    ")\n",
    "updated_pipeline.state"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "62c32a8a-c5fb-407f-99c5-4a24dcd756d3",
   "metadata": {},
   "source": [
    "## Clean-up\n",
    "\n",
    "Delete the pipelines and spaces we created:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "36c22a1d-7f6d-4d62-9c31-2588e1c73e48",
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline.delete()\n",
    "space.delete()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bf5e515e-b30d-4d6a-926b-2ca94de837b6",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}