Add endpoint to receive webhook for actinia status update

.dockerignore

@@ -0,0 +1 @@
+docker

.flake8

@@ -10,4 +10,5 @@ per-file-ignores =
     ./tests/cloudevent_receiver_server.py: E501
     # remove when implemented:
     ./src/actinia_cloudevent_plugin/api/cloudevent.py: E501
-    ./src/actinia_cloudevent_plugin/core/processing.py: F841, E501
+    ./src/actinia_cloudevent_plugin/api/hooks.py: F841
+    ./src/actinia_cloudevent_plugin/core/cloudevents.py: F841, E501

.vscode/launch.json

@@ -0,0 +1,24 @@
+{
+    "configurations": [
+        {
+            "name": "Docker: Python - Flask",
+            "type": "docker",
+            "request": "launch",
+            "preLaunchTask": "docker-run: debug",
+            "python": {
+                "pathMappings": [
+                    {
+                        "localRoot": "${workspaceFolder}",
+                        "remoteRoot": "/src/actinia-cloudevent-plugin"
+                    }
+                ],
+                "projectType": "flask"
+            },
+            "dockerServerReadyAction": {
+              "action": "openExternally",
+              "pattern": "Running on (https?://\\S+|[0-9]+)",
+              "uriFormat": "%s://localhost:%s/"
+            }
+        }
+    ]
+}

.vscode/tasks.json

@@ -0,0 +1,58 @@
+{
+	"version": "2.0.0",
+	"tasks": [
+		{
+			"type": "docker-build",
+			"label": "docker-build",
+			"platform": "python",
+			"dockerBuild": {
+				"tag": "actinia-cloudevent-plugin:latest",
+				"dockerfile": "${workspaceFolder}/docker/Dockerfile",
+				"context": "${workspaceFolder}"
+			}
+		},
+		{
+			"type": "docker-run",
+			"label": "docker-run: debug",
+			"dependsOn": [
+				"docker-build"
+			],
+			"python": {
+				"module": "flask",
+				"args": [
+					"run",
+					"--no-debugger",
+					"--host",
+					"0.0.0.0",
+					"--port",
+					"3003"
+				]
+			},
+			"dockerRun": {
+				"remove": true,
+				"network": "actinia-docker_actinia-dev",
+				"ports": [
+					{
+						"containerPort": 3003,
+						"hostPort": 3003
+					}
+				],
+				"env": {
+					"PYTHONUNBUFFERED": "1",
+					"PYTHONDONWRITEBYTECODE": "1",
+					"FLASK_APP": "actinia_cloudevent_plugin.main",
+					"FLASK_DEBUG": "1",
+					"FLASK_ENV": "development"
+				},
+				"customOptions": "--ip 172.18.0.12",
+				"volumes": [
+					{
+						"localPath": "${workspaceFolder}",
+						"containerPath": "/src/actinia-cloudevent-plugin",
+						"permissions": "rw"
+					}
+				]
+			}
+		}
+	]
+}

docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM alpine:3.21
+FROM alpine:3.22
 
 # python3 + pip3
 # hadolint ignore=DL3018
@@ -9,23 +9,17 @@ RUN /usr/bin/python -m venv --system-site-packages --without-pip /opt/venv
 # hadolint ignore=DL3013
 RUN python -m ensurepip && pip3 install --no-cache-dir --upgrade pip pep517 wheel
 
-# gunicorn
 # hadolint ignore=DL3013
-RUN pip3 install --no-cache-dir gunicorn
-
-# needed for tests
-# hadolint ignore=DL3013
-RUN pip3 install --no-cache-dir setuptools pwgen==0.8.2.post0 pytest==8.3.5 pytest-cov==6.0.0
+RUN pip3 install --no-cache-dir gunicorn && \
+    # needed for tests
+    pip3 install --no-cache-dir setuptools pwgen==0.8.2.post0 pytest==8.3.5 pytest-cov==6.0.0
 
 COPY . /src/actinia-cloudevent-plugin/
-
-# SETUPTOOLS_SCM_PRETEND_VERSION is only needed if in the plugin folder is no
-# .git folder
-ENV SETUPTOOLS_SCM_PRETEND_VERSION=0.0
+RUN pip3 install --no-cache-dir -e /src/actinia-cloudevent-plugin/ && \
+    # For tests:
+    chmod a+x /src/actinia-cloudevent-plugin/tests_with_cloudevent_receiver.sh
 
 WORKDIR /src/actinia-cloudevent-plugin
-RUN pip3 install --no-cache-dir -e .
-
-# For tests:
-RUN chmod a+x tests_with_cloudevent_receiver.sh && make install
 # RUN make test
+
+CMD ["gunicorn", "-b", "0.0.0.0:8088", "-w", "8", "--access-logfile=-", "-k", "gthread", "actinia_cloudevent_plugin.main:flask_app"]

docker/docker-compose.yml

@@ -11,4 +11,4 @@ services:
       - SYS_PTRACE
     ports:
       - "5000:5000"
-    network_mode: "host"
+    # network_mode: "host"

src/actinia_cloudevent_plugin/api/cloudevent.py

@@ -27,7 +27,7 @@
 from requests.exceptions import ConnectionError  # noqa: A004
 
 from actinia_cloudevent_plugin.apidocs import cloudevent
-from actinia_cloudevent_plugin.core.processing import (
+from actinia_cloudevent_plugin.core.cloudevents import (
     cloud_event_to_process_chain,
     receive_cloud_event,
     send_binary_cloud_event,

src/actinia_cloudevent_plugin/api/hooks.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python
+"""Copyright (c) 2025 mundialis GmbH & Co. KG.
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+Hello World class
+"""
+
+__license__ = "GPLv3"
+__author__ = "Carmen Tawalika"
+__copyright__ = "Copyright 2025 mundialis GmbH & Co. KG"
+__maintainer__ = "mundialis GmbH & Co. KG"
+
+import json
+
+import requests
+from cloudevents.conversion import to_binary
+from cloudevents.http import CloudEvent
+from flask import jsonify, make_response, request
+from flask_restful_swagger_2 import Resource, swagger
+
+from actinia_cloudevent_plugin.apidocs import hooks
+from actinia_cloudevent_plugin.model.response_models import (
+    SimpleStatusCodeResponseModel,
+)
+from actinia_cloudevent_plugin.resources.config import EVENTRECEIVER
+
+
+class Hooks(Resource):
+    """Webhook handling."""
+
+    def get(self, source_name):
+        """Cloudevent get method: not allowed response."""
+        _source_name = source_name
+        res = jsonify(
+            SimpleStatusCodeResponseModel(
+                status=405,
+                message="Method Not Allowed",
+            ),
+        )
+        return make_response(res, 405)
+
+    def head(self, source_name):
+        """Cloudevent head method: return empty response."""
+        _source_name = source_name
+        return make_response("", 200)
+
+    @swagger.doc(hooks.describe_hooks_post_docs)
+    def post(self, source_name) -> SimpleStatusCodeResponseModel:
+        """Translate actinia webhook call to cloudevent.
+
+        This method is called by HTTP POST actinia-core webhook
+        """
+        # only actinia as source supported so far
+        if source_name != "actinia":
+            return make_response(
+                jsonify(
+                    SimpleStatusCodeResponseModel(
+                        status=400,
+                        message="Bad Request: Source name not 'actinia'",
+                    ),
+                ),
+                400,
+            )
+
+        postbody = request.get_json(force=True)
+
+        if type(postbody) is dict:
+            postbody = json.dumps(postbody)
+        elif not isinstance(postbody, str):
+            postbody = str(postbody)
+
+        resp = json.loads(postbody)
+        if "resource_id" not in resp:
+            return make_response(
+                jsonify(
+                    SimpleStatusCodeResponseModel(
+                        status=400,
+                        message="Bad Request: No resource_id found in request",
+                    ),
+                ),
+                400,
+            )
+
+        # TODO: define when to send cloudevent
+        status = resp["status"]
+        if status == "finished":
+            # TODO send cloudevent
+            pass
+        terminate_status = ["finished", "error", "terminated"]
+        if status in terminate_status:
+            # TODO send cloudevent
+            pass
+
+        # TODO: move to common function from core.cloudevents
+        url = EVENTRECEIVER.url
+        try:
+            attributes = {
+                "specversion": "1.0",
+                "source": "/actinia-cloudevent-plugin",
+                "type": "com.mundialis.actinia.process.status",
+                "subject": "nc_spm_08/PERMANENT",
+                "datacontenttype": "application/json",
+            }
+            data = {"actinia_job": resp}
+            event = CloudEvent(attributes, data)
+            headers, body = to_binary(event)
+            requests.post(url, headers=headers, data=body)
+        except ConnectionError as e:
+            return f"Connection ERROR when returning cloudevent: {e}"
+        except Exception() as e:
+            return f"ERROR when returning cloudevent: {e}"
+
+        res = jsonify(
+            SimpleStatusCodeResponseModel(
+                status=200,
+                message="Thank you for your update",
+            ),
+        )
+        return make_response(res, 200)

src/actinia_cloudevent_plugin/apidocs/hooks.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+"""Copyright (c) 2025 mundialis GmbH & Co. KG.
+
+This program is free software: you can redistribute it and/or modify
+it under the terms of the GNU General Public License as published by
+the Free Software Foundation, either version 3 of the License, or
+(at your option) any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+Apidocs for webhook endpoint
+"""
+
+__license__ = "GPLv3"
+__author__ = "Carmen Tawalika"
+__copyright__ = "Copyright 2025 mundialis GmbH & Co. KG"
+__maintainer__ = "mundialis GmbH & Co. KG"
+
+
+from actinia_cloudevent_plugin.model.response_models import (
+    SimpleStatusCodeResponseModel,
+)
+
+describe_hooks_post_docs = {
+    # "summary" is taken from the description of the get method
+    "tags": ["cloudevent"],
+    "description": (
+        "Receives webhook with status update e.g. from actinia-core,"
+        " transforms to cloudevent and sends it to configurable endpoint."
+    ),
+    "responses": {
+        "200": {
+            "description": (
+                "This response returns a cloud event, "
+                "generated from actinia-core status"
+            ),
+            "schema": SimpleStatusCodeResponseModel,
+        },
+        "400": {
+            "description": "This response returns an error message",
+            "schema": SimpleStatusCodeResponseModel,
+        },
+    },
+}

src/actinia_cloudevent_plugin/endpoints.py

@@ -23,41 +23,15 @@
 __maintainer__ = "mundialis GmbH & Co. KG"
 
 
-import sys
-
-import werkzeug
-from flask import current_app, send_from_directory
 from flask_restful_swagger_2 import Api
 
 from actinia_cloudevent_plugin.api.cloudevent import Cloudevent
-from actinia_cloudevent_plugin.resources.logging import log
+from actinia_cloudevent_plugin.api.hooks import Hooks
 
 
 # endpoints loaded if run as actinia-core plugin as well as standalone app
 def create_endpoints(flask_api: Api) -> None:
     """Create plugin endpoints."""
-    app = flask_api.app
     apidoc = flask_api
-
-    package = sys._getframe().f_back.f_globals["__package__"]  # noqa: SLF001
-    if package != "actinia_core":
-
-        @app.route("/")
-        def index():
-            try:
-                return current_app.send_static_file("index.html")
-            except werkzeug.exceptions.NotFound:
-                log.debug("No index.html found. Serving backup.")
-                # when actinia-cloudevent-plugin is installed in single mode,
-                # the swagger endpoint would be "latest/api/swagger.json".
-                # As api docs exist in single mode,
-                # use this fallback for plugin mode.
-                return """<h1 style='color:red'>actinia-metadata-plugin</h1>
-                    <a href="api/v1/swagger.json">API docs</a>"""
-
-        @app.route("/<path:filename>")
-        def static_content(filename):
-            # WARNING: all content from folder "static" will be accessible!
-            return send_from_directory(app.static_folder, filename)
-
     apidoc.add_resource(Cloudevent, "/")
+    apidoc.add_resource(Hooks, "/hooks/<string:source_name>")

src/actinia_cloudevent_plugin/main.py

@@ -42,7 +42,7 @@
 apidoc = Api(
     flask_app,
     title="actinia-cloudevent-plugin",
-    prefix=URL_PREFIX,
+    # prefix=URL_PREFIX,
     api_version=API_VERSION,
     api_spec_url=f"{URL_PREFIX}/swagger",
     schemes=["https", "http"],
@@ -53,8 +53,7 @@
                    """,
 )
 
-create_endpoints(apidoc)
-
+create_endpoints(flask_api=apidoc)
 
 if __name__ == "__main__":
     # call this for development only with: