ngits-drf-iot-client

Identity-agnostic Django/DRF client for the IoT ESB automation and command APIs. It lets any application (a single-tenant app, a multi-tenant panel, ...) act as a thin BFF over the shared IoT platform without duplicating domain logic — the same way ngits-drf-cube-proxy sits in front of analytics.

The automation domain (models, evaluation, command dispatch) lives in the IoT platform services. This package is only the transport + tenant-scoping glue used by the presentation layer.

Install

pip install ngits-drf-iot-client
# during local development, before it is published:
pip install -e ../ngits-drf-iot-client

Configure (host settings.py)

IOT_ESB_BASE_URL = "https://iot-esb.example.com"  # client appends /api/v1
IOT_ESB_TOKEN = env("IOT_ESB_TOKEN")              # sent as X-Token
IOT_CLIENT_IDENTITY_RESOLVER = "myapp.iot.resolve_scope"
# optional:
IOT_CLIENT_TIMEOUT = 10

The resolver maps an authenticated user to the tenant(s) they may access:

# myapp/iot.py  (single-tenant app)
def resolve_scope(user):
    tenant = get_tenant_for(user)                    # your own lookup
    return None if tenant is None else tenant.uuid   # None -> 403

# multi-tenant panel
def resolve_scope(user):
    return list(get_tenants_for(user))               # iterable of tenant UUIDs

Use in a DRF view

from rest_framework import viewsets
from rest_framework.response import Response
from iot_client import IotClientMixin

class AutomationViewSet(IotClientMixin, viewsets.ViewSet):
    def list(self, request):
        client = self.get_iot_client()
        return Response(client.automations.list(tenants=self.get_tenant_scope()))

    def create(self, request):
        client = self.get_iot_client()
        return Response(client.automations.create(request.data), status=201)

Client surface

IotEsbClient exposes one sub-client per ESB resource:

• client.automations — list, get, create, update, delete, activate, deactivate, rule_types, actions;
• client.alert_rules — list, get, create, update, delete, activate, deactivate, condition_kinds, list_executions, get_execution, ack_execution;
• client.commands — create, get;
• client.event_subscriptions — list, upsert, update, delete.

All read methods accept a tenants scope; errors raise IotClientError(status_code=...).

Alert-rule executions

client.alert_rules.list_executions(tenants=scope, acked=False, **{"from": since})
client.alert_rules.get_execution(execution_id, tenants=scope)
client.alert_rules.ack_execution(
    execution_id, acked_by=f"myapp:user:{user.pk}", comment="on it", tenants=scope
)

acked_by is an opaque, host-assigned identifier; the ESB treats a repeated ack as a no-op. The notification policy of a rule (notify_on_triggered, notify_on_resolved, reminder_after_minutes) is sent through the regular create / update payloads.

Receiving platform events (alert notifications)

The IoT platform does not notify people itself: it delivers domain events (alert.triggered, alert.resolved, alert.reminder, alert.acknowledged) to the hosts that subscribed to them, and each host decides who gets notified.

1. Settings

INSTALLED_APPS += ["iot_client"]          # for the management command

IOT_CLIENT = {
    "EVENTS_SECRET": env("IOT_EVENTS_SECRET"),        # HMAC key, shared with the platform
    "EVENTS_CONSUMER": "myapp",                       # host name in the subscription key
    "TENANT_UUIDS_PROVIDER": "myapp.iot.tenant_uuids",  # () -> iterable of tenant UUIDs
    # optional, default: all four alert.* types
    # "EVENT_TYPES": ["alert.triggered", "alert.resolved"],
}

2. Receiving endpoint

Each event is a JSON POST with the headers X-Event-Id, X-Event-Type, X-Event-Timestamp (unix seconds) and X-Event-Signature: sha256=<hex HMAC-SHA256(secret, "{timestamp}.{body}")>. IotEventSignaturePermission verifies the signature in constant time and rejects timestamps older/newer than 300 s:

from rest_framework.response import Response
from rest_framework.views import APIView
from iot_client.events import IotEventSignaturePermission

class IotEventsView(APIView):
    authentication_classes = []            # signature only -> invalid = 403
    permission_classes = [IotEventSignaturePermission]

    def post(self, request):
        event = request.data               # schema_version, event_id, event_type, ...
        if already_processed(event["event_id"]):   # dedupe: delivery is at-least-once
            return Response(status=200)
        handle(event)                      # any exception -> 5xx -> platform retries
        return Response(status=200)

Deduplication by event_id and handling are the host's job. Without EVENTS_SECRET the permission raises IotClientImproperlyConfigured. For other setups use verify_signature(secret, body, timestamp, signature) directly, or override IotEventSignaturePermission.get_secret().

3. Registration (deploy step)

python manage.py iot_register_event_subscription --env prod \
    --target https://myapp.example.com/api/v1/iot/events/ [--tenant UUID ...]

For every tenant (--tenant, or all from TENANT_UUIDS_PROVIDER) it upserts one subscription keyed by (tenant, env, consumer) with the configured secret and EVENT_TYPES, so running it on every deploy never creates duplicates. It does not send is_active, so a subscription switched off on the platform stays off. The command exits non-zero if any tenant failed.