Flask Decorators

Overview

This is a sample Flask application implemented using the Flask Decorators in the Action Provider Toolkit. The Toolkit provides an ActionProviderBlueprint with five decorators which are used to decorate functions that will be run when the Action Provider is invoked. Each decorator corresponds to one of the Action Provider Interface endpoints. The decorators available are:

  • action_run

  • action_status

  • action_cancel

  • action_release

  • action_log

Using this Tool

The ActionProviderBlueprint is exactly like a Flask Blueprint, except it has been customized to implement the Action Provider Interface and ties together much of the tooling available in the rest of the Toolkit to provide a streamlined development experience. The ActionProviderBlueprint will:

  • Validate that incoming requests to your ActionProvider adhere to the ActionRequest schema

  • Validate that incoming requests to your ActionProvider adhere to your ActionProvider’s defined input-schema

  • Automatically create routes implementing the Action Provider Interface

  • Enforce that only expected users can acccess your ActionProvider’s introspection and run endpoints.

  • Return valid views to callers.

All the ActionProvider developer needs to do is create an ActionProviderDescription and use it when creating the ActionProviderBlueprint:

from globus_action_provider_tools.flask.apt_blueprint import (
    ActionProviderBlueprint)
from globus_action_provider_tools.data_types import (
    ActionProviderDescription)

description = ActionProviderDescription(...)
aptb = ActionProviderBlueprint(
    name="apt",
    import_name=__name__,
    url_prefix="/apt",
    provider_description=description,
)

Note

The ActionProviderBlueprint is really just a Flask Blueprint in disguise. As such, any keyword arguments you pass to it get passed onto the underlying Blueprint constructor, giving you the familiar interface and capabilities of the Flask ecosystem.

Once the ActionProviderBlueprint has been created, use its decorators to register functions which implement your ActionProvider’s logic:

@aptb.action_run
def my_action_run(action_request: ActionRequest, auth: AuthState):
    pass

@aptb.action_status
def my_action_status(action_id: str, auth: AuthState):
    pass

@aptb.action_cancel
def my_action_cancel(action_id: str, auth: AuthState):
    pass

@aptb.action_release
def my_action_release(action_id: str, auth: AuthState):
    pass

@aptb.action_log
def my_action_log(action_id: str, auth: AuthState):
    pass

Note

It’s required that your decorated functions accept two positional arguments with the correct types. For the action_run function, the argument types need to be an ActionRequest and an AuthState. The rest of the functions will have argument types of str and AuthState. Within your function, you will have access to the requestors Globus Authentication information.

The toolkit provides a convenient way of authorizing access to an Action when a request to view or modify the Action’s execution gets made. You can import them via:

from globus_action_provider_tools.authorization import (
    authorize_action_access_or_404, authorize_action_management_or_404)

To use these, obtain an ActionStatus object from your ActionProvider’s storage backend and use the provided AuthState argument:

@aptb.action_status
def my_action_status(action_id: str, auth: AuthState):
    # Lookup ActionStatus via action_id
    action_status = ...
    authorize_action_access_or_404(action_status, auth)
    ...

@aptb.action_cancel
def my_action_cancel(action_id: str, auth: AuthState):
    # Lookup ActionStatus via action_id
    action_status = ...
    authorize_action_management_or_404(action_status, auth)
    ...

Note

You generally only want to use authorize_action_access_or_404 in the action_status and action_log endpoint functions. action_cancel and action_release should use authorize_action_management_or_404.

Later, register the ActionProviderBlueprint to a Flask app exactly as you would register any other Flask Blueprint and run your ActionProvider:

from flask import Flask

app = Flask(__name__)
app.config.from_object("config")
app.register_blueprint(aptb)
app.run()

Note

One important difference between the ActionProviderBlueprint and a regular Flask Blueprint is that internally, the ActionProviderBlueprint will create a TokenChecker instance upon registration with a Flask application. This TokenChecker is what handles all authentication and authorization to the ActionProvider. As such, the Flask application must be configured to contain a valid Globus Auth client ID and client secret. An ActionProviderBlueprint will attempt to pull these credentials from application it is registered to’s configuration. First, the Blueprint checks to see if there are configuration keys of the form “BLUEPRINT_NAME_CLIENT_ID” and “BLUEPRINT_NAME_CLIENT_SECRET”. If those configuration keys are not found, the Blueprint will look for the keys “CLIENT_ID” and “CLIENT_SECRET” in the app’s configuration. If these configuration values cannot be found, the Action Provider will not be able to authenticate requests against Globus Auth.

As an example, if we created the following ActionProviderBlueprint:

aptb = ActionProviderBlueprint(
    name="apt",
    import_name=__name__,
    url_prefix="/apt",
    provider_description=description,
)

Once aptb gets registered with a Flask app, it will attempt to find the “APT_CLIENT_ID” and “APT_CLIENT_SECRET” keys in the Flask application’s configuration. Failing to find those, it will search for and use the Flask application’s “CLIENT_ID” and “CLIENT_SECRET” values.

Example Configuration

To run this example Action Provider, you need to generate your own CLIENT_ID, CLIENT_SECRET, and SCOPE. It may be useful to follow the directions for generating each of these located at Set Up an Action Provider in Globus Auth. Once you have those three values, place the CLIENT_ID and CLIENT_SECRET into the example Action Provider’s config.py and update the ActionProviderDescription’s globus_auth_scope value in blueprint.py.

We recommend creating a virtualenvironment to install project dependencies and run the Action Provider. Once the virtualenvironment has been created and activated, run the following:

cd examples/apt_blueprint
pip install -r requirements.txt
python app.py

Action Provider Implementation

Configuring the Flask Application: config.py
CLIENT_ID = ""
CLIENT_SECRET = ""
Creating the Flask Application: app.py
import logging

from flask import Flask

from examples.apt_blueprint import config
from examples.apt_blueprint.blueprint import aptb


def create_app():
    app = Flask(__name__)
    app.config.from_object(config)
    app.register_blueprint(aptb)
    app.logger.setLevel(logging.INFO)
    return app


if __name__ == "__main__":
    app = create_app()
    app.run(debug=True)
Creating the ActionProviderBlueprint: blueprint.py
from datetime import datetime, timezone
from typing import Dict, List, Set

from flask import request
from pydantic import BaseModel, Field

from examples.apt_blueprint.backend import simple_backend
from globus_action_provider_tools.authorization import (
    authorize_action_access_or_404,
    authorize_action_management_or_404,
)
from globus_action_provider_tools.data_types import (
    ActionProviderDescription,
    ActionRequest,
    ActionStatus,
    ActionStatusValue,
    AuthState,
)
from globus_action_provider_tools.exceptions import ActionConflict, ActionNotFound
from globus_action_provider_tools.flask import ActionProviderBlueprint
from globus_action_provider_tools.flask.types import ActionLogReturn, ActionStatusReturn

from .backend import simple_backend


class ActionProviderInput(BaseModel):
    utc_offset: int = Field(
        ..., title="UTC Offset", description="An input value to this ActionProvider"
    )

    class Config:
        schema_extra = {"example": {"utc_offset": 10}}


description = ActionProviderDescription(
    globus_auth_scope="https://auth.globus.org/scopes/d3a66776-759f-4316-ba55-21725fe37323/action_all",
    title="What Time Is It Right Now?",
    admin_contact="support@whattimeisrightnow.example",
    synchronous=True,
    input_schema=ActionProviderInput,
    api_version="1.0",
    subtitle="Another exciting promotional tie-in for whattimeisitrightnow.com",
    description="",
    keywords=["time", "whattimeisitnow", "productivity"],
    visible_to=["public"],
    runnable_by=["all_authenticated_users"],
    administered_by=["support@whattimeisrightnow.example"],
)


aptb = ActionProviderBlueprint(
    name="apt",
    import_name=__name__,
    url_prefix="/apt",
    provider_description=description,
)


@aptb.action_enumerate
def action_enumeration(auth: AuthState, params: Dict[str, Set]) -> List[ActionStatus]:
    """
    This is an optional endpoint, useful for allowing requestors to enumerate
    actions filtered by ActionStatus and role.

    The params argument will always be a dict containing the incoming request's
    validated query arguments. There will be two keys, 'statuses' and 'roles',
    where each maps to a set containing the filter values for the key. A typical
    params object will look like:

        {
            "statuses": {<ActionStatusValue.ACTIVE: 3>},
            "roles": {"creator_id"}
        }

    Notice that the value for the "statuses" key is an Enum value.
    """
    statuses = params["statuses"]
    roles = params["roles"]
    matches = []

    for _, action in simple_backend.items():
        if action.status in statuses:
            # Create a set of identities that are allowed to access this action,
            # based on the roles being queried for
            allowed_set = set()
            for role in roles:
                identities = getattr(action, role)
                if isinstance(identities, str):
                    allowed_set.add(identities)
                else:
                    allowed_set.update(identities)

            # Determine if this request's auth allows access based on the
            # allowed_set
            authorized = auth.check_authorization(allowed_set)
            if authorized:
                matches.append(action)

    return matches


@aptb.action_run
def my_action_run(action_request: ActionRequest, auth: AuthState) -> ActionStatusReturn:
    """
    Implement custom business logic related to instantiating an Action here.
    Once launched, collect details on the Action and create an ActionStatus
    which records information on the instantiated Action and gets stored.
    """
    action_status = ActionStatus(
        status=ActionStatusValue.ACTIVE,
        creator_id=str(auth.effective_identity),
        label=action_request.label or None,
        monitor_by=action_request.monitor_by or auth.identities,
        manage_by=action_request.manage_by or auth.identities,
        start_time=str(datetime.now(tz=timezone.utc)),
        completion_time=None,
        release_after=action_request.release_after or "P30D",
        display_status=ActionStatusValue.ACTIVE,
        details={},
    )
    simple_backend[action_status.action_id] = action_status
    return action_status


@aptb.action_status
def my_action_status(action_id: str, auth: AuthState) -> ActionStatusReturn:
    """
    Query for the action_id in some storage backend to return the up-to-date
    ActionStatus. It's possible that some ActionProviders will require querying
    an external system to get up to date information on an Action's status.
    """
    action_status = simple_backend.get(action_id)
    if action_status is None:
        raise ActionNotFound(f"No action with {action_id}")
    authorize_action_access_or_404(action_status, auth)
    return action_status


@aptb.action_cancel
def my_action_cancel(action_id: str, auth: AuthState) -> ActionStatusReturn:
    """
    Only Actions that are not in a completed state may be cancelled.
    Cancellations do not necessarily require that an Action's execution be
    stopped. Once cancelled, the ActionStatus object should be updated and
    stored.
    """
    action_status = simple_backend.get(action_id)
    if action_status is None:
        raise ActionNotFound(f"No action with {action_id}")

    authorize_action_management_or_404(action_status, auth)
    if action_status.is_complete():
        raise ActionConflict("Cannot cancel complete action")

    action_status.status = ActionStatusValue.FAILED
    action_status.display_status = f"Cancelled by {auth.effective_identity}"
    simple_backend[action_id] = action_status

    return action_status


@aptb.action_release
def my_action_release(action_id: str, auth: AuthState) -> ActionStatusReturn:
    """
    Only Actions that are in a completed state may be released. The release
    operation removes the ActionStatus object from the data store. The final, up
    to date ActionStatus is returned after a successful release.
    """
    action_status = simple_backend.get(action_id)
    if action_status is None:
        raise ActionNotFound(f"No action with {action_id}")

    authorize_action_management_or_404(action_status, auth)
    if not action_status.is_complete():
        raise ActionConflict("Cannot release incomplete Action")

    action_status.display_status = f"Released by {auth.effective_identity}"
    simple_backend.pop(action_id)
    return action_status


@aptb.action_log
def my_action_log(action_id: str, auth: AuthState) -> ActionLogReturn:
    """
    Action Providers can optionally support a logging endpoint to return
    detailed information on an Action's execution history. Pagination and
    filters are supported as query parameters and can be used to control what
    details are returned to the requestor.
    """
    pagination = request.args.get("pagination")
    filters = request.args.get("filters")
    return {
        "time": "TODAY",
        "code": 200,
        "description": f"This is an example of a detailed log entry for {action_id}",
        "details": {
            "action_id": "Transfer",
            "filters": filters,
            "pagination": pagination,
        },
    }