helin-edge-sdk 0.0.2

Receive and respond to Helin Platform edge-module-requests.

helin-edge-sdk

The Helin Edge SDK allows you to securely and reliably get and set JSON payloads via the Helin Platform's edge-module-requests API: get_configuration, set_configuration and get_health to your containers (modules) deployed on Nodes on the edge.

For more information see https://docs.helinplatform.com

Install

pip install helin-edge-sdk

Usage

Implement the three methods and run the client:

from typing import Any
from helin_edge_sdk import EdgeModuleRequestsHandler, EdgeModuleClient, ConfigurationFailedError

class MyModule(EdgeModuleRequestsHandler):
    def __init__(self) -> None:
        self.configuration = {"opcua_ip": "123.34.233.1", "opcua_port": 4840, "poll_interval_ms": 1000}

    def on_get_configuration(self, payload: dict[str, Any]) -> Any:
        return self.configuration

    def on_set_configuration(self, payload: dict[str, Any]) -> Any:
        cfg = payload.get("configuration")
        if cfg is None:
            raise ConfigurationFailedError("missing configuration", status_code=400,
                                           current_configuration=self.configuration)
        self.configuration = cfg
        return {"configuration": cfg}

    def on_get_health(self, payload: dict[str, Any]) -> Any:
        return {"metrics": [{"name": "opcua_connected", "value": 1}]}

EdgeModuleClient.from_edge_environment(MyModule()).run() #Fetches all required env variables on your provisioned Node

Each callback receives the decompressed request payload and returns a JSON-serializable value, which is wrapped as {"status": 200, "response": <value>}

compressed, and published back to the caller.

Raise ConfigurationFailedError to return the configuration-failure envelope:

{"status": <code>, "response": {"error": ..., "current_configuration": ...}}.

Status codes

Handler outcome	Status
Returns a value	200
Raises ValueError	400
Raises ConnectionError	400
Unknown method	400
Raises FileNotFoundError	404
Raises ConfigurationFailedError	its status_code (default 409)
Raises any other exception	500

Local testing

# 1. start a plaintext broker
docker compose -f docker/docker-compose.test.yml up -d

# 2. run the module against it
python -m examples.run_local

# 3. invoke a method (simulates the Platform API)
python -m examples.publish_method get_health
python -m examples.publish_method set_configuration '{"configuration": {"opcua_ip": "10.0.0.5", "opcua_port": 4840}}'

Develop

poetry install
poetry run pytest -m "not integration"   # unit tests, no broker
poetry run pytest                        # all tests (needs the broker above)
poetry run ruff check
poetry run ty check

Architecture

flowchart TB
    subgraph platform["Cloud"]
        API["Platform API <br/> POST /api/v1/edge-module-requests/nodes/{node_id}/modules/{module_id} <br/> GET /api/v1/edge-module-requests/nodes/{node_id}/modules/{module_id} <br/> GET /api/v1/edge-module-requests/nodes/{node_id}/modules/{module_id}/health"]
    end

    subgraph hub["Azure IoT Hub (Cloud)"]
        IOTHUB
    end

    subgraph edge["Node — Edge runtime"]
        EDGEHUB["edgeHub<br/>MQTT broker :8883 (TLS)"]
        WL["Edge workload API<br/>unix:///…/workload.sock<br/>/sign · /trust-bundle"]

        subgraph mod["Module using helin-edge-sdk"]
            direction TB
            CLIENT["EdgeModuleClient<br/>from_edge_environment()"]
            LISTENER["MqttModuleListener<br/>(paho-mqtt transport)"]
            HANDLER["EdgeModuleRequestsHandler<br/>(your on_* callbacks)"]
            CLIENT --> LISTENER
            LISTENER -- "dispatch(method, payload)" --> HANDLER
            HANDLER -- "result / ConfigurationFailedError" --> LISTENER
        end
    end

    API <== "edge-module-requests<br/>(gzip payload)" ==> IOTHUB
    IOTHUB <== "cloud uplink" ==> EDGEHUB

    EDGEHUB -- "SUB - (gzip payload)" --> LISTENER
    LISTENER -- "PUB - (gzip payload)" --> EDGEHUB

    LISTENER -- "SAS token + trust bundle" --> WL