Flask Blueprint¶
globus-action-provider-tools
provides a custom
Flask Blueprint object
with decorators for registering functions which implement Action Provider
interfaces.
Overview¶
This is a sample Flask application implemented using the Flask Blueprint 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 a Flask Blueprint which has been customized to implement the Action Provider Interface.
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 access 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 requester’s 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¶
CLIENT_ID = ""
CLIENT_SECRET = ""
import logging
from flask import Flask
from examples.apt_blueprint import config
from examples.apt_blueprint.blueprint import aptb
from globus_action_provider_tools.flask.helpers import assign_json_provider
def create_app():
app = Flask(__name__)
assign_json_provider(app)
app.logger.setLevel(logging.DEBUG)
app.config.from_object(config)
app.register_blueprint(aptb)
return app
if __name__ == "__main__":
app = create_app()
app.run(debug=True)
from datetime import datetime, timezone
from typing import Dict, List, Set
from flask import request
from pydantic import BaseModel, Field
from globus_action_provider_tools import (
ActionProviderDescription,
ActionRequest,
ActionStatus,
ActionStatusValue,
AuthState,
)
from globus_action_provider_tools.authorization import (
authorize_action_access_or_404,
authorize_action_management_or_404,
)
from globus_action_provider_tools.flask import ActionProviderBlueprint
from globus_action_provider_tools.flask.exceptions import ActionConflict, ActionNotFound
from globus_action_provider_tools.flask.types import (
ActionCallbackReturn,
ActionLogReturn,
)
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 requesters 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
) -> ActionCallbackReturn:
"""
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=datetime.now(timezone.utc).isoformat(),
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) -> ActionCallbackReturn:
"""
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) -> ActionCallbackReturn:
"""
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) -> ActionCallbackReturn:
"""
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 requester.
"""
pagination = request.args.get("pagination")
filters = request.args.get("filters")
return ActionLogReturn(
code=200,
description=f"This is an example of a detailed log entry for {action_id}",
**{
"time": "TODAY",
"details": {
"action_id": "Transfer",
"filters": filters,
"pagination": pagination,
},
},
)