Action provider code merge

.github/workflows/deployment.yml

@@ -4,9 +4,9 @@ on:
   push:
     branches:
       - main
-  pull_request:
-    branches:
-      - main
+  # pull_request:
+    # branches:
+      # - main
   workflow_dispatch:
     inputs:
       service_names:

README.md

@@ -19,3 +19,4 @@ TODOs
 - action provider basic tests
 
 - add docs along the way
+- octopus icon

action_provider/main.py

@@ -1,26 +1,102 @@
-"""Diaspora Action Provider."""
+"""Flask application for the Diaspora Action Provider."""
 
 from __future__ import annotations
 
 import os
 
+from flask import Blueprint
 from flask import Flask
+from globus_action_provider_tools import ActionProviderDescription
+from globus_action_provider_tools import ActionRequest
+from globus_action_provider_tools import ActionStatusValue
+from globus_action_provider_tools import AuthState
+from globus_action_provider_tools.flask import add_action_routes_to_blueprint
+from globus_action_provider_tools.flask.helpers import assign_json_provider
+from globus_action_provider_tools.flask.types import ActionCallbackReturn
 
 from action_provider import __version__
+from action_provider.utils import build_action_status
+from action_provider.utils import load_schema
 from common.utils import EnvironmentChecker
 
-app = Flask(__name__)
+CLIENT_ID = os.environ['CLIENT_ID']
+CLIENT_SECRET = os.environ['CLIENT_SECRET']
+CLIENT_SCOPE = os.environ['CLIENT_SCOPE']
+DEFAULT_SERVERS = os.environ['DEFAULT_SERVERS']
 
 
-@app.route('/')
-def hello_world() -> str:
-    """One and only entry point."""
-    version = __version__
-    client_id = os.environ['CLIENT_ID']
-    return (
-        'Hello World! from Action Provider '
-        f'version {version} with CLIENT_ID={client_id}'
+def action_run(
+    request: ActionRequest,
+    auth: AuthState,
+) -> ActionCallbackReturn:
+    """Produce or consume events."""
+    status = build_action_status(
+        auth,
+        ActionStatusValue.SUCCEEDED,
+        request,
+        result={'result': 'succeeded'},
     )
+    return status, 200
+    # action = request.body['action']
+    # if action == 'produce':
+    #     return action_produce(request, auth)
+    # else:
+    #     return action_consume(request, auth)
+
+
+def action_status(action_id: str, auth: AuthState) -> ActionCallbackReturn:
+    """Placeholder status endpoint."""
+    status = build_action_status(auth)
+    return status, 200
+
+
+def action_cancel(action_id: str, auth: AuthState) -> ActionCallbackReturn:
+    """Placeholder cancel endpoint."""
+    status = build_action_status(auth)
+    return status
+
+
+def action_release(action_id: str, auth: AuthState) -> ActionCallbackReturn:
+    """Placeholder release endpoint."""
+    status = build_action_status(auth)
+    return status
+
+
+def create_app() -> Flask:
+    """Create the Flask application."""
+    app = Flask(__name__)
+    assign_json_provider(app)
+    app.url_map.strict_slashes = False
+
+    skeleton_blueprint = Blueprint('diaspora', __name__)
+
+    provider_description = ActionProviderDescription(
+        globus_auth_scope=CLIENT_SCOPE,
+        title='Diaspora Action Provider',
+        admin_contact='haochenpan@uchicago.edu',
+        administered_by=['Diaspora Team', 'Globus Labs'],
+        api_version=__version__,
+        synchronous=True,
+        input_schema=load_schema(),
+        log_supported=False,
+        visible_to=['public'],
+    )
+
+    add_action_routes_to_blueprint(
+        blueprint=skeleton_blueprint,
+        client_id=CLIENT_ID,
+        client_secret=CLIENT_SECRET,
+        client_name=None,
+        provider_description=provider_description,
+        action_run_callback=action_run,
+        action_status_callback=action_status,
+        action_cancel_callback=action_cancel,
+        action_release_callback=action_release,
+    )
+
+    app.register_blueprint(skeleton_blueprint)
+
+    return app
 
 
 if __name__ == '__main__':
@@ -32,4 +108,5 @@ def hello_world() -> str:
         'CLIENT_SCOPE',
         'DEFAULT_SERVERS',
     )
-    app.run(host='0.0.0.0', port=8000)
+    app = create_app()
+    app.run(host='0.0.0.0', port=8000, debug=True)

action_provider/schema.json

@@ -0,0 +1,83 @@
+{
+    "title": "Diaspora Event Fabric Messaging Schema",
+    "type": "object",
+    "properties": {
+        "action": {
+            "description": "The action to perform: 'produce' for publishing events, 'consume' for retrieving recent events.",
+            "type": "string",
+            "enum": [
+                "produce",
+                "consume"
+            ]
+        },
+        "topic": {
+            "description": "The topic to publish or retrieve the events.",
+            "type": "string"
+        },
+        "msgs": {
+            "type": "array",
+            "description": "List of events, each formatted as a JSON string. Required for 'produce' action.",
+            "items": {
+                "type": "object"
+            }
+        },
+        "keys": {
+            "oneOf": [
+                {
+                    "type": "string",
+                    "description": "Optional single event key for 'produce' action."
+                },
+                {
+                    "type": "array",
+                    "description": "Optional list of event keys for 'produce' action.",
+                    "items": {
+                        "type": "string"
+                    }
+                }
+            ]
+        },
+        "ts": {
+            "description": "Timestamp in milliseconds since the beginning of the epoch to start retrieving messages from. Required for 'consume' action.",
+            "type": "integer",
+            "minimum": 0
+        },
+        "servers": {
+            "description": "Optional list of diaspora servers separated by commas (dev use only).",
+            "type": "string"
+        }
+    },
+    "additionalProperties": false,
+    "required": [
+        "action",
+        "topic"
+    ],
+    "dependencies": {
+        "action": {
+            "oneOf": [
+                {
+                    "properties": {
+                        "action": {
+                            "const": "produce"
+                        },
+                        "msgs": {
+                            "minItems": 1
+                        }
+                    },
+                    "required": [
+                        "msgs"
+                    ]
+                },
+                {
+                    "properties": {
+                        "action": {
+                            "const": "consume"
+                        },
+                        "ts": {
+                            "type": "integer"
+                        }
+                    }
+                }
+            ]
+        }
+    }
+}

action_provider/utils.py

@@ -0,0 +1,60 @@
+"""Diaspora Action Provider utilities.
+
+This module contains utility functions and classes for handling AWS MSK tokens,
+Kafka operations, and building action statuses.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import os
+from typing import Any
+
+from globus_action_provider_tools import ActionRequest
+from globus_action_provider_tools import ActionStatus
+from globus_action_provider_tools import ActionStatusValue
+from globus_action_provider_tools import AuthState
+
+
+def load_schema() -> dict[str, Any]:
+    """Load Event Schema."""
+    with open(
+        os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            'schema.json',
+        ),
+    ) as f:
+        schema = json.load(f)
+    return schema
+
+
+def build_action_status(
+    auth: AuthState,
+    status_value: ActionStatusValue | None = None,
+    request: ActionRequest | None = None,
+    result: dict[str, Any] | None = None,
+) -> ActionStatus:
+    """Build an ActionStatus object depending on whetherrequest is None."""
+    if request is None:
+        return ActionStatus(
+            status=ActionStatusValue.SUCCEEDED,
+            creator_id=auth.effective_identity,
+            start_time=str(datetime.datetime.now().isoformat()),
+            completion_time=str(datetime.datetime.now().isoformat()),
+            release_after='P30D',
+            display_status=ActionStatusValue.SUCCEEDED,
+            details={'result': None},
+        )
+    else:
+        return ActionStatus(
+            status=status_value,
+            creator_id=auth.effective_identity,
+            monitor_by=request.monitor_by,
+            manage_by=request.manage_by,
+            start_time=str(datetime.datetime.now().isoformat()),
+            completion_time=str(datetime.datetime.now().isoformat()),
+            release_after=request.release_after or 'P30D',
+            display_status=status_value,
+            details=result,
+        )

mkdocs.yml

@@ -52,9 +52,9 @@ theme:
       toggle:
         icon: material/brightness-4
         name: Switch to system preference
-  # favicon: static/favicon.png
+  # favicon: static/octopus.png
   # icon:
-  #   logo: logo
+    # logo: static/octopus.png
 
 watch:
   - mkdocs.yml

testing/__init__.py

@@ -0,0 +1,3 @@
+"""Testing code for Diaspora Service."""
+
+from __future__ import annotations

testing/fixtures.py

@@ -0,0 +1,22 @@
+"""Fixtures for Diaspora Service."""
+
+from __future__ import annotations
+
+import pytest
+
+from action_provider.main import create_app
+from testing.globus import get_access_token
+
+
+@pytest.fixture(scope='module')
+def client():
+    """Create the Flask service."""
+    app = create_app()
+    with app.test_client() as client:
+        yield client
+
+
+@pytest.fixture(scope='module')
+def access_token():
+    """Retrieve the access token."""
+    return get_access_token()

testing/globus.py

@@ -0,0 +1,32 @@
+"""Globus related testing code for Diaspora Service."""
+
+from __future__ import annotations
+
+import os
+
+from globus_sdk import ConfidentialAppAuthClient
+
+from common.utils import EnvironmentChecker
+
+
+def get_access_token() -> str:
+    """Get an access token to SERVER_CLIENT_ID."""
+    EnvironmentChecker.check_env_variables(
+        'SERVER_CLIENT_ID',
+        'SERVER_SECRET',
+        'CLIENT_SCOPE',
+    )
+
+    client_id = os.getenv('SERVER_CLIENT_ID')
+    client_secret = os.getenv('SERVER_SECRET')
+    requested_scopes = os.getenv('CLIENT_SCOPE')
+
+    ca = ConfidentialAppAuthClient(
+        client_id=client_id,
+        client_secret=client_secret,
+    )
+    token_response = ca.oauth2_client_credentials_tokens(
+        requested_scopes=requested_scopes,
+    )
+    access_token = token_response.by_resource_server[client_id]['access_token']
+    return access_token

tests/__init__.py

@@ -1,3 +1,3 @@
-"""Testing code for Diaspora Service."""
+"""Tests for Diaspora Service."""
 
 from __future__ import annotations