Flow_with_Compute.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "a95a8a30-8bdd-4d8f-83a2-945ea52ff693",
   "metadata": {},
   "source": [
    "# Flow with a Compute Function\n",
    "\n",
    "This notebook demonstrates how to construct a Globus Flow that includes a Globus Compute function call. Our flow will **scan a folder** in a Globus collection and **ingest the list of files** into a new entry in a search index. We'll use Globus Compute to **prepare the folder listing** for ingest into the index.\n",
    "\n",
    "The outline of the notebook is as follows.\n",
    "\n",
    "1. Import required Python modules and specify the search index and compute endpoint to be used.\n",
    "2. Authenticate with Globus and create Globus SDK FlowsClient object.\n",
    "3. **Define the function** and **register it** with Globus Compute (to get a UUID).\n",
    "4. **Define and deploy** the Flow.\n",
    "5. **Run the flow** with a Globus collection and path.\n",
    "\n",
    "Each time the flow is run, a new entry will be created in the search index. We'll use a **simple web application** to display the contents of the index so we can see entries appear as tutorial participants run their flows!"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "0aa7abbb-2158-48bb-b54f-9d8d5d0ba412",
   "metadata": {},
   "source": [
    "## Important: Join the Tutorial Users group!\n",
    "\n",
    "Before using this notebook, make sure you've joined the \n",
    "[Tutorial Users](https://app.globus.org/groups/50b6a29c-63ac-11e4-8062-22000ab68755/about) \n",
    "group! \n",
    "\n",
    "* Click the link above and then click \"Join this Group.\" "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "4a757973-5f87-4920-8c30-4a6ee78708d3",
   "metadata": {},
   "source": [
    "## 1. Import modules and specify Search Index and Compute Endpoint\n",
    "\n",
    "NOTE: The `source_collection` is one of our Globus tutorial collections. Because it's a mapped collection, it requires a special scope during authentication. Feel free to change this to any other mapped collection that you have access to. The flow will also work with any guest collection (aka \"shared collection\") without changing this setting."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "434ec760-bd75-4a2b-ad52-ed9c8ef7a98c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "import os\n",
    "import time\n",
    "import json\n",
    "import uuid\n",
    "import pickle\n",
    "import base64\n",
    "\n",
    "import globus_sdk\n",
    "import globus_sdk.scopes\n",
    "import globus_compute_sdk\n",
    "\n",
    "# ID of this tutorial notebook as registered with Globus Auth\n",
    "CLIENT_ID = 'f794186b-f330-4595-b6c6-9c9d3e903e47'\n",
    "\n",
    "# This search index will work for you, but free to replace it with your own search index\n",
    "search_index_id = \"2c635508-f599-4547-a0fc-536db83d1dee\"   # Created and shared by tutorial instructor\n",
    "\n",
    "# Feel free to replace the collection UUIDs below with those of your own collections\n",
    "source_collection = \"6c54cade-bde5-45c1-bdea-f4bd71dba2cc\"  # \"Globus Tutorial Collection 1\"\n",
    "\n",
    "# This compute endpoint will work for you, but feel free to replace it with your own endpoint\n",
    "compute_endpoint_id = '4b116d3c-1703-4f8f-9f6f-39921e5864df' # \"Globus Compute Tutorial Endpoint\""
   ]
  },
  {
   "cell_type": "markdown",
   "id": "d19c4931-fedf-4a01-b9c3-2a895fbe06d7",
   "metadata": {},
   "source": [
    "## 2. Authenticate and create a FlowClient and Globus Compute Client object\n",
    "\n",
    "If you're using this notebook with the Globus JupyterHub, the following code will get your login information from the JupyterHub and you won't have to login again. Otherwise, you'll be prompted to login to Globus.\n",
    "\n",
    "* We'll use the `flows_client` object (created by `globus_sdk.FlowsClient()`) to deploy our flow.\n",
    "* We'll use the `gcc` object (created by `globus_compute_sdk.Client()`) to register our Compute function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f6f2289d-48ed-41d2-8432-9db6e2e35db1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create transfer scope with data_access scope dependency from the source mapped collection.\n",
    "transfer_scope = globus_sdk.scopes.TransferScopes.make_mutable(\"all\")\n",
    "data_access_scope = globus_sdk.scopes.GCSCollectionScopeBuilder(source_collection).data_access\n",
    "transfer_scope.add_dependency(data_access_scope)\n",
    "\n",
    "# Get Globus Auth token data from the JupyterHub environment. If tokens already exist from logging into\n",
    "# jupyter.demo.globus.org, tokens from the environment can be used instead. Otherwise, do a Native App flow.\n",
    "globus_data_raw = os.getenv(\"GLOBUS_DATA\")\n",
    "if globus_data_raw:\n",
    "    tokens = pickle.loads(base64.b64decode(os.getenv('GLOBUS_DATA')))['tokens']\n",
    "else:\n",
    "    # Do a native app authentication flow to get tokens that allow us to interact with the Globus Flows service\n",
    "    scopes = [\n",
    "        \"openid\",\n",
    "        \"profile\",\n",
    "        \"email\",\n",
    "        transfer_scope,\n",
    "        globus_sdk.FlowsClient.scopes.manage_flows,\n",
    "        globus_sdk.FlowsClient.scopes.run_manage,\n",
    "    ]\n",
    "    native_auth_client = globus_sdk.NativeAppAuthClient(CLIENT_ID)\n",
    "    native_auth_client.oauth2_start_flow(requested_scopes=scopes)\n",
    "    print(f\"Login Here:\\n\\n{native_auth_client.oauth2_get_authorize_url()}\")\n",
    "    \n",
    "    # Authenticate and come back with your authorization code; paste it into the prompt below.\n",
    "    auth_code = input('Authorization Code: ')\n",
    "    response = native_auth_client.oauth2_exchange_code_for_tokens(auth_code)\n",
    "    \n",
    "    # Save the new token in a place where the flows client can retrieve it.\n",
    "    tokens = response.by_resource_server\n",
    "    \n",
    "    # These are the saved scopes for the flow\n",
    "    print(json.dumps(tokens, indent=2))\n",
    "\n",
    "# Uncomment the line below to introspect tokens\n",
    "#print(json.dumps(tokens, indent=2))\n",
    "\n",
    "# Create a variable for storing flow scope tokens. Each newly deployed flow scope needs to be authorized separately,\n",
    "# and will have its own set of tokens. Save each of these tokens by scope.\n",
    "saved_flow_scopes = {}\n",
    "\n",
    "# Add a callback to the flows client for fetching scopes. It will draw scopes from `saved_flow_scopes`\n",
    "def get_flow_authorizer(flow_id):\n",
    "    return globus_sdk.AccessTokenAuthorizer(access_token=saved_flow_scopes[flow_id]['access_token'])\n",
    "\n",
    "# Setup the Flow client, using tokens from our Jupyterhub login to access the Globus Flows service, and\n",
    "# set the `get_flow_authorizer` callback for any new flows we authorize.\n",
    "flows_authorizer = globus_sdk.AccessTokenAuthorizer(access_token=tokens['flows.globus.org']['access_token'])\n",
    "flows_client = globus_sdk.FlowsClient(authorizer=flows_authorizer)\n",
    "\n",
    "# Setup the Globus Compute client\n",
    "gcc = globus_compute_sdk.Client()     # get a Globus Compute client"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "17b8939d-3480-4b1b-ad30-da1e99c26541",
   "metadata": {},
   "source": [
    "## 3.a Define the Compute function\n",
    "\n",
    "We'll use the Transfer Action Provider's \"ls\" action to scan the contents of the folder. The output of the action looks like this:\n",
    "\n",
    "```\n",
    "    {\n",
    "        \"DATA\": [\n",
    "          {\n",
    "            \"DATA_TYPE\": \"file\",\n",
    "            \"group\": \"tutorial\",\n",
    "            \"last_modified\": \"2024-09-19 17:11:23+00:00\",\n",
    "            \"link_group\": null,\n",
    "            \"link_last_modified\": null,\n",
    "            \"link_size\": null,\n",
    "            \"link_target\": null,\n",
    "            \"link_user\": null,\n",
    "            \"name\": \"file1.txt\",\n",
    "            \"permissions\": \"0644\",\n",
    "            \"size\": 4,\n",
    "            \"type\": \"file\",\n",
    "            \"user\": \"tutorial\"\n",
    "          },\n",
    "          {\n",
    "            \"DATA_TYPE\": \"file\",\n",
    "            \"group\": \"tutorial\",\n",
    "            \"last_modified\": \"2024-09-19 17:11:23+00:00\",\n",
    "            \"link_group\": null,\n",
    "            \"link_last_modified\": null,\n",
    "            \"link_size\": null,\n",
    "            \"link_target\": null,\n",
    "            \"link_user\": null,\n",
    "            \"name\": \"file2.txt\",\n",
    "            \"permissions\": \"0644\",\n",
    "            \"size\": 4,\n",
    "            \"type\": \"file\",\n",
    "            \"user\": \"tutorial\"\n",
    "          }\n",
    "        ],\n",
    "        \"DATA_TYPE\": \"file_list\",\n",
    "        \"absolute_path\": \"/home/share/godata/\",\n",
    "        \"endpoint\": \"6c54cade-bde5-45c1-bdea-f4bd71dba2cc\",\n",
    "        \"length\": 2,\n",
    "        \"path\": \"/home/share/godata/\",\n",
    "        \"rename_supported\": true,\n",
    "        \"symlink_supported\": false,\n",
    "        \"total\": 2\n",
    "      }\n",
    "```\n",
    "We want to translate this into a much simpler format so it can be added to a search index. We want the search entry to look like this:\n",
    "\n",
    "```\n",
    "      {\n",
    "        \"path\": \"/home/share/godata/\",\n",
    "        \"file_list\": [\n",
    "          \"file1.txt\",\n",
    "          \"file2.txt\",\n",
    "        ]\n",
    "      }\n",
    "\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f9d1f140-94b1-440d-88e5-5a13bc848a57",
   "metadata": {},
   "outputs": [],
   "source": [
    "def simplify_file_list(**folder_contents):\n",
    "    import time    # so we can add a timestampt\n",
    "    files = []\n",
    "    for file in folder_contents[\"DATA\"]:\n",
    "        files.append(file[\"name\"])\n",
    "    entry = {\n",
    "        \"time_published\": time.asctime(),\n",
    "        \"path\": folder_contents[\"path\"],\n",
    "        \"file_list\": files\n",
    "    }\n",
    "    return entry"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "6b3f140b-0198-4c89-9115-9caa200509b9",
   "metadata": {},
   "source": [
    "## 3.b Register the function with Globus Compute\n",
    "\n",
    "Now we'll use our Globus Compute Client (`gcc`) to register the function with Globus Compute and get a UUID that we can use in our Flow definition.\n",
    "\n",
    "NOTE: When you execute this, you'll probably be asked to login to Globus and allow the notebook to use Globus Compute on your behalf."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7d62977a-e8c9-429f-b9ac-dd100fa7b3b4",
   "metadata": {},
   "outputs": [],
   "source": [
    "function_uuid = gcc.register_function(simplify_file_list)\n",
    "print(f\"function uuid = {function_uuid}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "c1d92ade-8d7c-446a-8e78-4b126de92282",
   "metadata": {},
   "source": [
    "## 4.a Define the Flow\n",
    "\n",
    "Now, we'll define our flow. It has three states:\n",
    "1. ListFiles (Globus Transfer action provider)\n",
    "2. CreateEntry (Globus Compute action provider with our function)\n",
    "3. AddEntry (Globus Search action provider)\n",
    "\n",
    "Pay attention to how the output of the `ListFiles` state gets passed into the `CreateEntry` state, and the output from `CreateEntry` gets passed to `AddEntry`. Every time the flow is run, each state adds its output to the run's state (`$`) and the following state takes what it needs for its input. When a run completes, you can inspect the inputs and outputs of each state in the Event Log tab to see the details!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cbfad11b-1ab7-445b-b989-f6fc04c7b50e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define flow\n",
    "flow_definition = {\n",
    "    \"Comment\": \"Add a new entry in a search index containing a summary of the contents of a folder\",\n",
    "    \"StartAt\": \"ListFiles\",\n",
    "    \"States\": {\n",
    "        \"ListFiles\": {\n",
    "            \"Comment\": \"List the contents of the folder specified by the flow inputs\",\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://transfer.actions.globus.org/ls\",\n",
    "            \"Parameters\": {\n",
    "                \"endpoint_id.$\": \"$.input.source.id\",\n",
    "                \"path.$\": \"$.input.source.path\"\n",
    "            },\n",
    "            \"ResultPath\": \"$.ls_result\",\n",
    "            \"Next\": \"CreateEntry\"\n",
    "        },\n",
    "        \"CreateEntry\": {\n",
    "            \"Comment\": \"Use Globus Compute to process the file listing and turn it into a search entry\",\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://compute.actions.globus.org\",\n",
    "            \"Parameters\": {\n",
    "                \"endpoint\": compute_endpoint_id,\n",
    "                \"function\": function_uuid,\n",
    "                \"kwargs.$\":  \"$.ls_result.details\"\n",
    "            },\n",
    "            \"ResultPath\": \"$.CreateEntry\",\n",
    "            \"Next\": \"AddEntry\"\n",
    "        },\n",
    "        \"AddEntry\": {\n",
    "            \"Comment\": \"Add the new search entry to the search index\",\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://actions.globus.org/search/ingest\",\n",
    "            \"Parameters\": {\n",
    "               \"search_index\": search_index_id,\n",
    "               \"subject.$\": \"$.input.entry_subject\",\n",
    "               \"visible_to\": [\"public\"],\n",
    "               \"content.$\": \"$.CreateEntry.details.results[0].output\"\n",
    "            },\n",
    "            \"ResultPath\": \"$.AddEntry\",\n",
    "            \"End\": True\n",
    "        }\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cae99cfb-fbae-4aa0-b97d-9aebdbc72ca4",
   "metadata": {},
   "source": [
    "## 4.b Define the input schema\n",
    "\n",
    "All Flows require schemas, written in JSON Schema. The Input Schema is deployed with the ``flow_definition``.\n",
    "\n",
    "When the flow is run, the input schema is used to:\n",
    "* display an input entry form in the Globus web app, and\n",
    "* validate that the flow's inputs are of the right types before proceeding."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fe6a4eed-d0a1-4f3f-942c-33464597311e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define input schema\n",
    "input_schema = {\n",
    "    \"required\": [\n",
    "        \"input\"\n",
    "    ],\n",
    "    \"properties\": {\n",
    "        \"input\": {\n",
    "            \"type\": \"object\",\n",
    "            \"required\": [\n",
    "                \"source\",\n",
    "                \"entry_subject\"\n",
    "            ],\n",
    "            \"properties\": {\n",
    "                \"source\": {\n",
    "                    \"type\": \"object\",\n",
    "                    \"title\": \"Select source collection and path\",\n",
    "                    \"description\": \"The source collection and path (path MUST end with a slash)\",\n",
    "                    \"format\": \"globus-collection\",\n",
    "                    \"required\": [\n",
    "                        \"id\",\n",
    "                        \"path\"\n",
    "                    ],\n",
    "                    \"properties\": {\n",
    "                        \"id\": {\n",
    "                            \"type\": \"string\",\n",
    "                            \"format\": \"uuid\"\n",
    "                        },\n",
    "                        \"path\": {\n",
    "                            \"type\": \"string\"\n",
    "                        }\n",
    "                    },\n",
    "                    \"additionalProperties\": False\n",
    "                },\n",
    "                \"entry_subject\": {\n",
    "                    \"type\": \"string\",\n",
    "                    \"title\": \"Subject of the entry to be created\",\n",
    "                    \"description\": \"Unique string for each entry in the search index\"\n",
    "                }\n",
    "            },\n",
    "            \"additionalProperties\": False\n",
    "        }\n",
    "    },\n",
    "    \"additionalProperties\": False\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dca25db0-e4f5-462c-8c4d-c2b16a786ca3",
   "metadata": {},
   "source": [
    "## 4.c Deploy the flow\n",
    "\n",
    "We'll use our `flows_client`, the `flow_definition`, and `input_schema` to deploy the flow. We'll create a random display name for the flow."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5d6d7b68-b020-4ba3-bf0e-5e7a28241d29",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the flow\n",
    "# Set the flow's title so you can easily identify it\n",
    "flow_title = f\"Tutorial-List-Compute-Search-{str(uuid.uuid4())[:6]}\"\n",
    "\n",
    "# Deploy the flow!\n",
    "flow = flows_client.create_flow(\n",
    "    title=flow_title,\n",
    "    definition=flow_definition,\n",
    "    input_schema=input_schema\n",
    ")\n",
    "flow_id = flow['id']\n",
    "\n",
    "# Add a dependency on the Transfer service, since we'll use it in the flow\n",
    "flow_scope = globus_sdk.SpecificFlowClient(flow_id).scopes.make_mutable(\"user\")\n",
    "flow_scope.add_dependency(transfer_scope)\n",
    "\n",
    "\"\"\"\n",
    "# If you change the flow, you will need to update it.\n",
    "# For example, to make this flow visibe to another user:\n",
    "flow = flows_client.update_flow(\n",
    "    flow_id=flow_id, \n",
    "    flow_viewers=[f\"urn:globus:auth:identity:{identity_id}\"]),\n",
    ")\n",
    "\"\"\"\n",
    "\n",
    "print(f\"Successfully created flow: '{flow_title}'\")\n",
    "print(f\"(ID: {flow_id})\")\n",
    "print(f\"Flow scope: {flow_scope}\\n\\n\")\n",
    "print(f\"View the flow in the Web App: https://app.globus.org/flows/{flow_id}\")\n",
    "print(f\"Note: You can start your flow directly from the Web App\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a15151d3-a1dc-4db1-b9c0-5c0d647d74ca",
   "metadata": {},
   "source": [
    "# 5. Run the flow\n",
    "\n",
    "It's time to run our flow! Remember, this flow requires three inputs: a *unique subject string* for the search entry we're going to create, and a *source collection* and *source path* to scan. We need to set those inputs up and then call the Flows API to run the flow."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "12723ced-680c-4cc9-b680-41be355b64c4",
   "metadata": {},
   "source": [
    "## 5.a Fetch user identity and create a unique subject string\n",
    "\n",
    "Every entry in a search index must have a unique subject string to identify it. We'll use our username and a random string to create it. We just logged in, so we can get the username from the access token for the Globus Auth API."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f516e636-a49f-4365-b70a-ba45a4bee3a1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create an Auth client so we can look up identities\n",
    "auth_authorizer = globus_sdk.AccessTokenAuthorizer(access_token=tokens['auth.globus.org']['access_token'])\n",
    "ac = globus_sdk.AuthClient(authorizer=auth_authorizer)\n",
    "\n",
    "# Get the user's primary identity\n",
    "primary_identity = ac.oauth2_userinfo()\n",
    "identity_id = primary_identity['sub']\n",
    "\n",
    "# Create a unique subject for the new search entry\n",
    "new_entry_subject = f\"{primary_identity['preferred_username']}/{str(uuid.uuid4())[:8]}\"\n",
    "\n",
    "print(f\"Entry's subject will be \\\"{ new_entry_subject }\\\".\")"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "3d1d08bd-6427-49bb-8579-f06b337dbeb4",
   "metadata": {},
   "source": [
    "## 5.b Define flow input\n",
    "\n",
    "If your flow includes parameterized input, you must provide values for those properties when running the flow. Like the flow definition, flow input is defined as a JSON document. You must provide a value for each input property in your flow. (Input properties are part of the flow's \"state\" and can be accessed in a flow definition by prefixing values with `$.` and providing the path to the property, as seen in the flow definition above).\n",
    "\n",
    "For the `MoveFiles` action, we must specify source and destination collection IDs and source and destination paths. For the `SetPermissions` action we must specify the collection ID, the type of entity to which we're granting permission, the entity's ID, and the permission (read or read/write)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c36b69df-beca-4e8f-9ec0-0afebba5ec51",
   "metadata": {},
   "outputs": [],
   "source": [
    "# We already defined the unique subject string above.\n",
    "# Now, we need the folder to be scanned. We need a source_collection and source_path.\n",
    "\n",
    "# Feel free to replace the collection UUIDs below with those of your own collections\n",
    "source_collection = \"6c54cade-bde5-45c1-bdea-f4bd71dba2cc\"  # \"Globus Tutorial Collection 1\"\n",
    "# The following path exists on Globus Tutorial Collection 1. \n",
    "# Replace with another path, maybe \"/\"?, if you change the source_collection.\n",
    "source_path = \"/home/share/godata\"   \n",
    "\n",
    "flow_input = {\n",
    "    \"input\": {\n",
    "        # Transfer input\n",
    "        \"source\": {\n",
    "            \"id\": source_collection,\n",
    "            \"path\": source_path\n",
    "        },\n",
    "        \"entry_subject\": new_entry_subject   # We just created this above, remember?\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "a47ee34b-c7ee-43b6-a058-bd5c77c1d8e3",
   "metadata": {},
   "source": [
    "## 5.c Authorize the flow\n",
    "\n",
    "Once your flow has been created, in order to run it, you will need to authorize it to interact with other services on your behalf. Globus Flow service generates a dedicated scope for each flow. To give consent to this flow, we need to get a properly scoped access token (see `flow_scope` above, where we deployed the flow), and then we can use this token to execute the flow. Note that you will be required to consent again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cbf3e02a-9037-4923-a618-c2f70d4fa326",
   "metadata": {},
   "outputs": [],
   "source": [
    "# If the flow scope is already saved, we don't need a new one.\n",
    "if flow_id not in tokens:\n",
    "    # Do a native app authentication flow and get tokens that include the newly deployed flow scope\n",
    "    native_auth_client = globus_sdk.NativeAppAuthClient(CLIENT_ID)\n",
    "    native_auth_client.oauth2_start_flow(requested_scopes=flow_scope)\n",
    "    print(f\"Login Here:\\n\\n{native_auth_client.oauth2_get_authorize_url()}\")\n",
    "    \n",
    "    # Authenticate and come back with your authorization code; paste it into the prompt below.\n",
    "    auth_code = input('Authorization Code: ')\n",
    "    token_response = native_auth_client.oauth2_exchange_code_for_tokens(auth_code)\n",
    "    \n",
    "    # Save the new token in a place where the flows client can retrieve it.\n",
    "    tokens[flow_id] = token_response.by_resource_server[flow_id]\n",
    "    \n",
    "    # These are the saved scopes for the flow\n",
    "    print(json.dumps(tokens, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "aaca2624-3a05-4f6c-a5dd-473e398db0df",
   "metadata": {},
   "source": [
    "## 5.d Run the flow\n",
    "\n",
    "We're finally ready to run the flow. You can monitor and manage your flow runs from the Globus Web App (https://app.globus.org/runs)\n",
    "\n",
    "Note: If you run the flow multiple times it will fail after the first run because, once the access rule is set on the collection/path, setting it again will fail. Run the code in the \"Remove Access Rule\" cell below to clear things up on the destination endpoint before running the flow again."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "72f4ac75-3b20-4151-945c-b6b3a14cee2d",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get a client for the flow\n",
    "specific_flow_authorizer = globus_sdk.AccessTokenAuthorizer(\n",
    "    access_token=tokens[flow_id]['access_token'],\n",
    ")\n",
    "specific_flow_client = globus_sdk.SpecificFlowClient(\n",
    "    flow_id=flow_id,\n",
    "    authorizer=specific_flow_authorizer,\n",
    ")\n",
    "\n",
    "# Run the flow\n",
    "# Set a descriptive label for this flow run\n",
    "run_label = f\"Scan/Compute/Search tutorial run for {primary_identity['preferred_username']}\"\n",
    "run = specific_flow_client.run_flow(\n",
    "  body=flow_input,\n",
    "  label=run_label,\n",
    "  tags=['tutorial', 'scan-compute-search-flow']\n",
    ")\n",
    "\n",
    "# Get run details\n",
    "run_id = run['run_id']\n",
    "run_status = run['status']\n",
    "print(\"This flow can be monitored in the Web App:\")\n",
    "print(f\"https://app.globus.org/runs/{run_id}\")\n",
    "print(f\"Flow run started with ID: {run_id} - Status: {run_status}\")\n",
    "\n",
    "# Poll the Flow service to check on the status of the flow\n",
    "while run_status == 'ACTIVE':\n",
    "    time.sleep(5)\n",
    "    run = flows_client.get_run(run_id)\n",
    "    run_status = run['status']\n",
    "    print(f'Run status: {run_status}')\n",
    "    \n",
    "# Run completed\n",
    "print(json.dumps(run.data, indent=2))"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e2dc323a-ac95-4c66-a691-adeba1011e70",
   "metadata": {},
   "source": [
    "## 6. View entries in the search index\n",
    "\n",
    "Our flow adds a new entry to the search index defined at the beginning of this notebook. We can search the index to see all of these entries! Here we generate a link that opens the [Argonne Community Data Co-op](https://acdc.alcf.anl.gov/) web application to view the search index and its entries."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0cee87a6-a9a7-45db-b0b8-21ce45e62448",
   "metadata": {},
   "outputs": [],
   "source": [
    "print(\"Click here to view the tutorial search index:\")\n",
    "print(f\"https://acdc.alcf.anl.gov/globus-tutorial/{search_index_id}/\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2bec1de3-076e-4a3e-a7e2-2b69f16ecc48",
   "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.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}