iot-edge-validator 1.0.6

Description:

iotedgevalidator 1.0.6

iot-edge-validator

This package is a wrapper around the azure-iot-device SDK to provide standardized exception handling and direct method request validation.
Official Documentation | Source code | Package PyPI
Table of Contents

Versioning
Getting Started

Prerequisites
Basic Examples



Versioning
This repository adheres to Semantic Versioning. It will be maintained through the CHANGELOG.md and in GitHub Releases. It's important to note that you must maintain the version with your releases in iot/edge/validator/_version.py, otherwise a new package version will fail to get published.
Getting Started
This section provides basic examples with the iot-edge-validator.
Prerequisites


Python 3.7 or later is required to use this package.


You must have an Azure subscription and Azure IoT Edge Device to use this package.


Basic Examples


Install via pip:
pip install iot-edge-validator



Import and say hello:
from iot.edge.validator import __version__


print(f"hello world from iot-edge-validator version: {__version__}")



Basic usage:
from typing import Any, Dict, Union

from azure.iot.device.iothub.models.methods import MethodRequest, MethodResponse

from iot.edge.validator import (
format_exception_error,
generate_error_response,
compare_dictionary,
)

EXPECTED_METHOD_NAME: str = "some_method_name"

EXPECTED_METHOD_PAYLOAD: Dict[str, Any] = {
"some": {},
"expected": {},
"payload": {},
}


def validate_method_requests(method_request: MethodRequest) -> Union[MethodResponse, None]:
"""validation handler for some_method_name listener"""
if method_request.name == EXPECTED_METHOD_NAME:
pass
else:
return generate_error_response(method_request,
f"received unknown method request for {method_request.name}",
400,
)


def validate_some_method_name_request(method_request: MethodRequest) -> Union[MethodResponse, None]:
"""validation for expected payload of some_method_name direct method request"""
# top level basic format validation
error_msg = compare_dictionary(
d1=method_request.payload,
d2=EXPECTED_METHOD_PAYLOAD,
value_match=False,
recurse=False,
)
if error_msg:
return generate_error_response(
method_request, f"error parsing payload: {error_msg}", 400
)
return None