Create & Run a Flow

These examples guide you through creating and running a flow using the Globus Flows service.

Note that users are restricted to only creating one flow unless covered by a Globus subscription. Therefore, these examples will also include an option to delete your flow.

Create and Delete Hello World Flow

This script provides commands to create, list, and delete your flows. The flow definition used for it is simple and baked into the script, but this demonstrates the minimal flow creation and deletion process.

manage_flow_minimal.py [download]

import argparse
import sys

import globus_sdk

# tutorial client ID
# we recommend replacing this with your own client for any production use-cases
CLIENT_ID = "61338d24-54d5-408f-a10d-66c06b59f6d2"


def create_flow(app: globus_sdk.GlobusApp, args: argparse.Namespace) -> None:
    with globus_sdk.FlowsClient(app=app) as flows_client:
        print(
            flows_client.create_flow(
                title=args.title,
                definition={
                    "StartAt": "DoIt",
                    "States": {
                        "DoIt": {
                            "Type": "Action",
                            "ActionUrl": "https://actions.globus.org/hello_world",
                            "Parameters": {
                                "echo_string": "Hello, Asynchronous World!",
                            },
                            "End": True,
                        }
                    },
                },
                input_schema={},
                subtitle="A flow created by the SDK tutorial",
            )
        )


def delete_flow(app: globus_sdk.GlobusApp, args: argparse.Namespace) -> None:
    with globus_sdk.FlowsClient(app=app) as flows_client:
        print(flows_client.delete_flow(args.flow_id))


def list_flows(app: globus_sdk.GlobusApp) -> None:
    with globus_sdk.FlowsClient(app=app) as flows_client:
        for flow in flows_client.list_flows(filter_roles="flow_owner"):
            print(f"title: {flow['title']}")
            print(f"id: {flow['id']}")
            print()


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("action", choices=["create", "delete", "list"])
    parser.add_argument("-f", "--flow-id", help="Flow ID for delete")
    parser.add_argument("-t", "--title", help="Name for create")
    args = parser.parse_args()

    with globus_sdk.UserApp("manage-flow-example", client_id=CLIENT_ID) as app:
        try:
            if args.action == "create":
                if args.title is None:
                    parser.error("create requires --title")
                create_flow(app, args)
            elif args.action == "delete":
                if args.flow_id is None:
                    parser.error("delete requires --flow-id")
                delete_flow(app, args)
            elif args.action == "list":
                list_flows(app)
            else:
                raise NotImplementedError()
        except globus_sdk.FlowsAPIError as e:
            print(f"API Error: {e.code} {e.message}")
            print(e.text)
            sys.exit(1)


if __name__ == "__main__":
    main()

Run a Flow

This next example is distinct. It runs a flow but has no capability to create or delete a flow. Note how SpecificFlowClient is used – this class allows users to access the flow-specific scope and provides the methods associated with that scope of running flows and resuming runs.

The login code is slightly different from the previous example, as it has to key off of the Flow ID in order to act appropriately.

run_flow_minimal.py [download]

import argparse
import sys

import globus_sdk

# tutorial client ID
# we recommend replacing this with your own client for any production use-cases
CLIENT_ID = "61338d24-54d5-408f-a10d-66c06b59f6d2"


def run_flow(app: globus_sdk.GlobusApp, args: argparse.Namespace) -> None:
    with globus_sdk.SpecificFlowClient(args.FLOW_ID, app=app) as flow_client:
        print(flow_client.run_flow({}))


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("FLOW_ID", help="Flow ID to run")
    args = parser.parse_args()

    with globus_sdk.UserApp("manage-flow-example", client_id=CLIENT_ID) as app:
        try:
            run_flow(app, args)
        except globus_sdk.FlowsAPIError as e:
            print(f"API Error: {e.code} {e.message}")
            print(e.text)
            sys.exit(1)


if __name__ == "__main__":
    main()

Create, Delete, and Run Flows

The following example combines the previous two. It has to further enhance the login code to account for the two different styles of login which it supports, but minimal other adjustments are needed.

Depending on the operation chosen, either the FlowsClient or the SpecificFlowClient will be used, and the login flow will also be appropriately parametrized.

manage_flow.py [download]

import argparse
import sys

import globus_sdk

# tutorial client ID
# we recommend replacing this with your own client for any production use-cases
CLIENT_ID = "61338d24-54d5-408f-a10d-66c06b59f6d2"


def create_flow(app: globus_sdk.GlobusApp, args: argparse.Namespace) -> None:
    with globus_sdk.FlowsClient(app=app) as flows_client:
        print(
            flows_client.create_flow(
                title=args.title,
                definition={
                    "StartAt": "DoIt",
                    "States": {
                        "DoIt": {
                            "Type": "Action",
                            "ActionUrl": "https://actions.globus.org/hello_world",
                            "Parameters": {
                                "echo_string": "Hello, Asynchronous World!",
                            },
                            "End": True,
                        }
                    },
                },
                input_schema={},
                subtitle="A flow created by the SDK tutorial",
            )
        )


def delete_flow(app: globus_sdk.GlobusApp, args: argparse.Namespace) -> None:
    with globus_sdk.FlowsClient(app=app) as flows_client:
        print(flows_client.delete_flow(args.flow_id))


def list_flows(app: globus_sdk.GlobusApp) -> None:
    with globus_sdk.FlowsClient(app=app) as flows_client:
        for flow in flows_client.list_flows(filter_roles="flow_owner"):
            print(f"title: {flow['title']}")
            print(f"id: {flow['id']}")
            print()


def run_flow(app: globus_sdk.GlobusApp, args: argparse.Namespace) -> None:
    with globus_sdk.SpecificFlowClient(args.flow_id, app=app) as flow_client:
        print(flow_client.run_flow({}))


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("action", choices=["logout", "create", "delete", "list", "run"])
    parser.add_argument("-f", "--flow-id", help="Flow ID for delete and run")
    parser.add_argument("-t", "--title", help="Name for create")
    args = parser.parse_args()

    with globus_sdk.UserApp("manage-flow-example", client_id=CLIENT_ID) as app:
        try:
            if args.action == "logout":
                app.logout(sweep=True)
            elif args.action == "create":
                if args.title is None:
                    parser.error("create requires --title")
                create_flow(app, args)
            elif args.action == "delete":
                if args.flow_id is None:
                    parser.error("delete requires --flow-id")
                delete_flow(app, args)
            elif args.action == "list":
                list_flows(app)
            elif args.action == "run":
                if args.flow_id is None:
                    parser.error("run requires --flow-id")
                run_flow(app, args)
            else:
                raise NotImplementedError()
        except globus_sdk.FlowsAPIError as e:
            print(f"API Error: {e.code} {e.message}")
            print(e.text)
            sys.exit(1)


if __name__ == "__main__":
    main()