Automation_Using_Flows_With_Search.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<img src=\"img/automation_using_flows_header.png\">\n",
    "\n",
    "In this notebook we demonstrate how the Globus Flow service can be used to automate data management at scale, with data discovery as a key aspect of the flow.\n",
    "\n",
    "We demonstrate a flow that automates a common design pattern: \n",
    "1. moving data from one system to another \n",
    "2. making the data accessible to collaborators \n",
    "3. publishing the metadata to a search index for discovery \n",
    "\n",
    "The Globus flow is illustrated below.\n",
    "\n",
    "<img src=\"img/transfer_set_permissions_ingest_flow.png\" alt=\"Transfer and set permissions flow\" align=\"CENTER\" style=\"width: 90%;\"/>\n",
    "\n",
    "This flow is often needed to manage data coming from instruments, e.g., image files can be moved from local storage attached to a microscope (acquisition machine) to a high-performance storage system where they may be accessed by all members of the research project. Metdata is gathered and pushed to a search index, with support for some metadata discoverable by anyone and other metadata available only to collaborators.\n",
    "\n",
    "We will walk through the following tasks:\n",
    "1. Authenticate with Globus and get tokens for accessing various services.\n",
    "1. Create a search index to publish metadata to, if it does not already exist.\n",
    "1. Define and register a flow with Globus that has the steps decribed above.\n",
    "1. Execute a flow using configurable inputs for the collections and the access permissions.\n",
    "\n",
    "In our example, we'll grant access to the [Tutorial Users group](https://app.globus.org/groups/50b6a29c-63ac-11e4-8062-22000ab68755/about))."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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",
    "\n",
    "# ID of this tutorial notebook as registered with Globus Auth\n",
    "CLIENT_ID = 'f794186b-f330-4595-b6c6-9c9d3e903e47'\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",
    "destination_collection = \"a6f165fa-aee2-4fe5-95f3-97429c28bf82\"  # \"Globus Tutorials on ALCF Eagle\"\n",
    "my_collaborators = \"50b6a29c-63ac-11e4-8062-22000ab68755\"  # \"Tutorial Users\" group"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# A. Authentication and Authorization\n",
    "\n",
    "All interactions between users and services on the Globus automation platform are governed by the Globus Auth service. In particular, this means that consent must be given by the user for each interaction taking place on their part, including in this notebook.\n",
    "\n",
    "The first time you interact with each service such as the Flows service, or even a flow instance, you will be provided a link to perform the consent flow. You must click the link to complete the consent flow which will launch in a new tab. When complete, copy the code string, return to the notebook, and  paste the code into the input box that is presented below the link to begin the flow.\n",
    "\n",
    "We will encounter authorization steps in a couple of places:\n",
    "1. When deploying a new flow on the Globus Flow service; deploying a flow requires (a) an identity that is associated with a Globus subscription, and (b) access to the Flow service scope.\n",
    "1. When executing a flow.\n",
    "\n",
    "Access to the Flows service is already granted to you by virtue of authenticating to the JupyterHub running this notebook. Note: If you're running this notebook in your own environment you will need to manually log into Globus Auth and get tokens using a native app authorization flow (see the `Platform_Introduction` notebook for an example of how to initiate this flow)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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.SearchClient.scopes.all,\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",
    "search_client = globus_sdk.SearchClient(authorizer=globus_sdk.AccessTokenAuthorizer(tokens['search.api.globus.org']['access_token']))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Fetch User Identity\n",
    "\n",
    "When transferring files to the guest collection we will put them in a directory named `<identity_id>-shared-files`, just to uniquely identify it from other directories. Let's fetch our user id for this purpose."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "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",
    "print(f\"Username: {primary_identity['preferred_username']} (ID: {identity_id})\")\n",
    "print(f\"Notifications will be sent to: {primary_identity['email']}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Get or Create Globus Search Index\n",
    "\n",
    "If you already have a trial index, the code below will retrieve it. Otherwise, a new trial index will be created for use in this tutorial. You can also replace `search_index` with any other Globus Search index ID."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "index_name = \"tutorial-notebook-index\"\n",
    "\n",
    "# Fetch all indices to which the user has access\n",
    "indices = [si for si in search_client.get(\"/v1/index_list\").data['index_list']\n",
    "           if si['is_trial'] \n",
    "           and si['display_name'] == index_name\n",
    "           and 'owner' in si['permissions']\n",
    "          ]\n",
    "\n",
    "# If an index was found with the criteria above, re-use it\n",
    "# Otherwise, create a new index\n",
    "if indices:\n",
    "    tutorial_index = indices[0]\n",
    "    print('Found existing trial index!')\n",
    "else:\n",
    "    tutorial_index = search_client.create_index(\n",
    "        display_name=index_name,\n",
    "        description=\"A trial index for use in Globus tutorials\"\n",
    "    ).data\n",
    "    print('Index created for use in Globus tutorials.')\n",
    "\n",
    "search_index = tutorial_index['id']\n",
    "\n",
    "print(tutorial_index['display_name'])\n",
    "print(tutorial_index['description'])\n",
    "print(search_index)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# B. Flow Deployment\n",
    "\n",
    "## Define a flow\n",
    "\n",
    "* Flows are composed of *action* invocations.\n",
    "* Each action invocation reads from and contributes back to the flow *state* which can be accessed in flow steps using the `InputPath` and `ResultPath` properties of an Action.\n",
    "* Actions are specified with an `ActionUrl` property. The `ActionUrl` is the address of an *action provider*: An API that provides actions you can invoke with your flow. Globus provides a number of action providers for Globus services, and you can also create your own using the [Action Provider Tools package](https://action-provider-tools.readthedocs.io/en/latest/).\n",
    "* Each action provider defines its accepted input schema specifying the permitted input format. For example, the Globus Transfer action provider requires source and destination collection IDs as well as source and destination paths.\n",
    "* Actions are linked via their `Next` property. The last action in a flow sets the `End` property to `true`.\n",
    "\n",
    "\n",
    "Our flow includes the following actions: `MoveFiles`, `SetPermission` and `SearchIngest`. The same Action `SearchIngest` is used in two steps to add metadata with varying visibility."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define flow\n",
    "flow_definition = {\n",
    "    \"Comment\": \"Transfer files to guest collection, set access permissions and publish metadata in search\",\n",
    "    \"StartAt\": \"TransferFiles\",\n",
    "    \"States\": {\n",
    "        \"TransferFiles\": {\n",
    "            \"Comment\": \"Transfer to a guest collection\",\n",
    "            # https://docs.globus.org/api/flows/hosted-action-providers/ap-transfer-transfer/\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://actions.automate.globus.org/transfer/transfer\",\n",
    "            \"Parameters\": {\n",
    "                \"source_endpoint_id.$\": \"$.input.source.id\", \n",
    "                \"destination_endpoint_id.$\": \"$.input.destination.id\",\n",
    "                \"transfer_items\": [\n",
    "                    {\n",
    "                        \"source_path.$\": \"$.input.source.path\",\n",
    "                        \"destination_path.$\": \"$.input.destination.path\",\n",
    "                        \"recursive.$\": \"$.input.recursive_tx\"\n",
    "                    }\n",
    "                ],\n",
    "            },\n",
    "            \"ResultPath\": \"$.TransferFiles\",\n",
    "            \"WaitTime\": 60,\n",
    "            \"Next\": \"SetPermission\"\n",
    "        },\n",
    "        \"SetPermission\": {\n",
    "            \"Comment\": \"Grant read permission on the data to a Globus user or group\",\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://actions.automate.globus.org/transfer/set_permission\",\n",
    "            # https://docs.globus.org/api/flows/hosted-action-providers/ap-transfer-set-permission/\n",
    "            \"Parameters\": {\n",
    "                \"endpoint_id.$\": \"$.input.destination.id\",\n",
    "                \"path.$\": \"$.input.destination.path\",\n",
    "                \"operation\": \"CREATE\",\n",
    "                \"permissions\": \"r\",\n",
    "                \"principal_type.$\": \"$.input.principal_type\",\n",
    "                \"principal.$\": \"$.input.principal_identifier\"\n",
    "            },\n",
    "            \"ExceptionOnActionFailure\": False,\n",
    "            \"ResultPath\": \"$.SetPermission\",\n",
    "            \"Next\": \"SearchIngest\"\n",
    "        },\n",
    "        \"SearchIngest\": {\n",
    "            \"Comment\": \"Ingest metadata to a Globus Search index\",\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://actions.globus.org/search/ingest\",\n",
    "            # https://docs.globus.org/api/flows/hosted-action-providers/ap-search-ingest/\n",
    "            \"Parameters\": {\n",
    "                \"search_index.$\": \"$.input.search_ingest_document.search_index\",  \n",
    "                \"subject.$\": \"$.input.search_ingest_document.search_subject\",  \n",
    "                \"visible_to.$\": \"$.input.search_ingest_document.search_visible_to\",  \n",
    "                \"content.$\": \"$.input.search_ingest_document.search_content_metadata\",\n",
    "                \"id.$\": \"$.input.search_ingest_document.search_entry_id\"\n",
    "            },\n",
    "            \"ResultPath\": \"$.SearchIngest\",\n",
    "            \"Next\": \"SearchIngestRestricted\"\n",
    "        },\n",
    "        \"SearchIngestRestricted\": {\n",
    "            \"Comment\": \"Ingest metadata with restricted visibility to a Globus Search index\",\n",
    "            \"Type\": \"Action\",\n",
    "            \"ActionUrl\": \"https://actions.globus.org/search/ingest\",\n",
    "            \"Parameters\": {\n",
    "                \"search_index.$\": \"$.input.search_ingest_document.search_index\",\n",
    "                \"subject.$\": \"$.input.search_ingest_document.search_subject\",\n",
    "                \"visible_to.$\": \"$.input.search_ingest_document.search_restricted_visible_to\",\n",
    "                \"content.$\": \"$.input.search_ingest_document.search_content_restricted_metadata\",\n",
    "                \"id.$\": \"$.input.search_ingest_document.search_restricted_entry_id\"\n",
    "            },\n",
    "            \"ResultPath\": \"$.SearchIngestRestricted\",\n",
    "            \"End\": True\n",
    "        }\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Define a schema\n",
    "\n",
    "* All Flows require schemas to validate user input is correct. \n",
    "* Flow Input Schemas are written in JSON Schema. \n",
    "* Input Schemas are deployed with the ``flow_definition`` and are checked when any user tries to run a fluw.\n",
    "\n",
    "Include a schema for our four states above:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define input schema\n",
    "input_schema = {\n",
    "    \"additionalProperties\": False,\n",
    "    \"required\": [\n",
    "        \"input\"\n",
    "    ],\n",
    "    \"properties\": {\n",
    "        \"input\": {\n",
    "            \"type\": \"object\",\n",
    "            \"required\": [\n",
    "                \"source\",\n",
    "                \"destination\",\n",
    "                \"recursive_tx\",\n",
    "                \"principal_type\",\n",
    "                \"principal_identifier\",\n",
    "                \"search_ingest_document\", \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",
    "                            \"default\": source_collection\n",
    "                        },\n",
    "                        \"path\": {\n",
    "                            \"type\": \"string\",\n",
    "                        }\n",
    "                    },\n",
    "                    \"additionalProperties\": False\n",
    "                },\n",
    "                \"destination\": {\n",
    "                    \"type\": \"object\",\n",
    "                    \"title\": \"Select destination collection and path\",\n",
    "                    \"description\": \"The destination collection and path (path MUST end with a slash); default collection is 'Globus Tutorials on ALCF Eagle'\",\n",
    "                    \"format\": \"globus-collection\",\n",
    "                    \"required\": [\n",
    "                        \"id\",\n",
    "                        \"path\"\n",
    "                    ],\n",
    "                    \"properties\": {\n",
    "                        \"id\": {\n",
    "                            \"type\": \"string\",\n",
    "                            \"format\": \"uuid\",\n",
    "                            \"default\": destination_collection\n",
    "                        },\n",
    "                        \"path\": {\n",
    "                            \"type\": \"string\",\n",
    "                            \"default\": f\"/automation-tutorial/{identity_id}-shared-files/\"\n",
    "                        }\n",
    "                    },\n",
    "                    \"additionalProperties\": False\n",
    "                },\n",
    "                \"recursive_tx\": {\n",
    "                    \"type\": \"boolean\",\n",
    "                    \"title\": \"Recursive transfer\",\n",
    "                    \"description\": \"Whether or not to transfer recursively, must be true when transferring a directory.\",\n",
    "                    \"default\": True,\n",
    "                },\n",
    "                \"principal_type\": {\n",
    "                    \"type\": \"string\",\n",
    "                    \"title\": \"Type of principal to share with\",\n",
    "                    \"description\": \"Specifies whether files are being shared with a user ('identity') or a group ('group'); default is 'group'\",\n",
    "                    \"enum\": [\n",
    "                        \"identity\",\n",
    "                        \"group\"\n",
    "                    ],\n",
    "                    \"default\": \"group\"\n",
    "                },\n",
    "                \"principal_identifier\": {\n",
    "                    \"type\": \"string\",\n",
    "                    \"title\": \"UUID of user identity or group\",\n",
    "                    \"description\": \"The user or group id to share with; default is 'Tutorial Users' group.\",\n",
    "                    \"format\": \"uuid\",\n",
    "                    \"default\": my_collaborators\n",
    "                },\n",
    "                \"search_ingest_document\": {\n",
    "                    \"type\": \"object\",\n",
    "                    \"title\": \"Search ingest document\",\n",
    "                    \"description\": \"The subject and metadata to be ingested into Globus Search\",\n",
    "                    \"required\": [\n",
    "                        \"search_index\",\n",
    "                        \"search_subject\",\n",
    "                        \"search_entry_id\",\n",
    "                        \"search_visible_to\",\n",
    "                        \"search_content_metadata\",\n",
    "                        \"search_restricted_entry_id\",\n",
    "                        \"search_restricted_visible_to\",\n",
    "                        \"search_content_restricted_metadata\"\n",
    "                    ],\n",
    "                    \"properties\": {\n",
    "                        \"search_index\": {\n",
    "                            \"type\": \"string\",\n",
    "                            \"title\": \"Globus Search Index ID\",\n",
    "                            \"description\": \"The UUID of the Globus Search index that will hold this metadata\",\n",
    "                            \"default\": search_index\n",
    "                        },\n",
    "                        \"search_subject\": {\n",
    "                            \"type\": \"string\",\n",
    "                            \"title\": \"Globus Search Subject\",\n",
    "                            \"description\": \"The target data subject for this index entry\"\n",
    "                        },\n",
    "                        \"search_entry_id\": {\n",
    "                            \"type\": \"string\",\n",
    "                            \"title\": \"Entry ID\",\n",
    "                            \"description\": \"Unique id for the public block of metadata\"\n",
    "                        },\n",
    "                        \"search_visible_to\": {\n",
    "                            \"type\": \"array\",\n",
    "                            \"title\": \"Visible To\",\n",
    "                            \"default\": [\"public\"],\n",
    "                            \"description\": \"Globus user(s) or group(s) that can view this metadata\"\n",
    "                        },\n",
    "                        \"search_content_metadata\": {\n",
    "                            \"type\": \"object\",\n",
    "                            \"title\": \"Search Metadata\",\n",
    "                            \"description\": \"The metadata for this index entry\",\n",
    "                            \"additionalProperties\": True\n",
    "                        },\n",
    "                        \"search_restricted_entry_id\": {\n",
    "                            \"type\": \"string\",\n",
    "                            \"title\": \"Restricted Entry ID\",\n",
    "                            \"description\": \"Unique id for the restricted block of metadata\"\n",
    "                        },\n",
    "                        \"search_restricted_visible_to\": {\n",
    "                            \"type\": \"array\",\n",
    "                            \"title\": \"Visible To (for restricted metadata)\",\n",
    "                            \"description\": \"Globus user(s) or group(s) that can view this restricted metadata\"\n",
    "                        },\n",
    "                        \"search_content_restricted_metadata\": {\n",
    "                            \"type\": \"object\",\n",
    "                            \"title\": \"Restricted Search Metadata\",\n",
    "                            \"description\": \"The restricted metadata for this index entry\",\n",
    "                            \"additionalProperties\": True\n",
    "                        }\n",
    "                    }\n",
    "                }\n",
    "            },\n",
    "            \"additionalProperties\": False\n",
    "        }\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create the flow\n",
    "In order to run this flow, we use the definition to create a flow in the Globus Flows service. In addition to the definition we created above, we will provide a unique title for the flow to make it easier to identify. If deployment succeeds, Flows will return the ID of your new flow."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create the flow\n",
    "# Set the flow's title so you can easily identify it\n",
    "flow_title = f\"Tutorial-Transfer-Publish-{str(uuid.uuid4())[:4]}\"\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",
    "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",
   "metadata": {},
   "source": [
    "# C. Flow Execution\n",
    "\n",
    "## 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), 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,
   "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",
   "metadata": {},
   "source": [
    "## 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,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Define flow inputs\n",
    "destination_path = f\"/automation-tutorial/{identity_id}-shared-files/\"\n",
    "flow_input = {\n",
    "    \"input\": {\n",
    "        # Transfer input\n",
    "        \"source\": {\n",
    "            \"id\": source_collection,\n",
    "            \"path\": \"/home/share/godata/\",\n",
    "        },\n",
    "        \"destination\": {\n",
    "            \"id\": destination_collection,\n",
    "            \"path\": destination_path,\n",
    "        },\n",
    "        \n",
    "        \"recursive_tx\": True,\n",
    "        # Grant access to the Tutorial Users group\n",
    "        \"principal_identifier\": my_collaborators,\n",
    "        \"principal_type\": \"group\",\n",
    "\n",
    "        # We could also grant access to a specific user, using their Globus identity ID\n",
    "        #\"principal\": identity_id,\n",
    "        #\"principal_type\": \"identity\",\n",
    "        \n",
    "        \"search_ingest_document\": {\n",
    "            \"search_index\": search_index,\n",
    "            \"search_subject\": f\"globus://{destination_collection}/{destination_path}\",\n",
    "            \"search_entry_id\" : \"Public Entry 001\",\n",
    "            \"search_visible_to\": [\"public\"],\n",
    "            \"search_content_metadata\": {\n",
    "                \"title\": \"Tutorial Destination and Path\",\n",
    "                \"organization\": \"My Research Organization\"\n",
    "            },\n",
    "            \"search_restricted_entry_id\" : \"Restricted Entry 001\",\n",
    "            \"search_restricted_visible_to\": [f\"urn:globus:auth:identity:{identity_id}\"],\n",
    "            \"search_content_restricted_metadata\": {\n",
    "                \"secret\": \"Formula 20220730\",\n",
    "                \"pathogen\":\"XLKFT-34895\",\n",
    "                \"antidote\": \"Unavailable\"\n",
    "            }\n",
    "        }\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 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,
   "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\"Transfer/Publish 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', 'transfer-publish-flow']\n",
    ")\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",
   "metadata": {},
   "source": [
    "## View the files on the guest collection\n",
    "Our files were moved to the guest collection and read access was granted to the Tutorial Users group. Members of the group can now access the files via the web app, CLI or the Globus APIs. Here we generate a link that opens the Globus web app file manager to view the collection."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from urllib.parse import urlencode, urlunsplit\n",
    "query_params = {\n",
    "    \"origin_id\": destination_collection,\n",
    "    \"origin_path\": destination_path\n",
    "}\n",
    "url = urlunsplit((\"https\", \"app.globus.org\", \"file-manager\", urlencode(query_params), ''))\n",
    "print(f\"View your files in the Globus web app here:\\n{url}\\n\\n\")\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Remove Access Rule\n",
    "\n",
    "You can remove the access permission directly, using the Globus SDK. And we may as well clean up the directory while we're at it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get the ID of the access rule from the flow action's output\n",
    "access_rule_id = run['details']['output']['SetPermission']['details']['access_id']\n",
    "\n",
    "transfer_authorizer = globus_sdk.AccessTokenAuthorizer(tokens['transfer.api.globus.org']['access_token'])\n",
    "tc = globus_sdk.TransferClient(authorizer=transfer_authorizer)\n",
    "\n",
    "# Remove the access rule\n",
    "response = tc.delete_endpoint_acl_rule(destination_collection, access_rule_id)\n",
    "print(response)\n",
    "\n",
    "# Delete the directory on the guest collection\n",
    "# DeleteData() automatically gets a submission_id for once-and-only-once submission\n",
    "label = \"Automation tutorial cleanup\"\n",
    "delete_data = globus_sdk.DeleteData(tc, destination_collection, label=label, recursive=True)\n",
    "\n",
    "## Recursively delete the destination path contents (given recursive flag set above)\n",
    "delete_data.add_item(destination_path)\n",
    "tc.endpoint_autoactivate(destination_collection)\n",
    "submit_result = tc.submit_delete(delete_data) \n",
    "print(f\"DELETE Task ID: {submit_result['task_id']}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Review Permissions\n",
    "\n",
    "Ensure the permissions no longer exist:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"View sharing permissions: \\nhttps://app.globus.org/file-manager/collections/{destination_collection}/sharing\")"
   ]
  }
 ],
 "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}