add iot hub features

Makefile

@@ -94,3 +94,14 @@ docs-serve: ## serve documentation
 
 .PHONY: ci-test-docs
 ci-test-docs: docs ## run CI test for documentation
+
+# ---
+# Project
+# ---
+
+.PHONY: server
+server: ## run server
+	poetry run uvicorn workshop_azure_iot.core:app \
+		--host 0.0.0.0 \
+		--port 8000 \
+		--reload

iot_hub.env.template

@@ -0,0 +1 @@
+IOT_HUB_DEVICE_CONNECTION_STRING="CHANGE_ME"

pyproject.toml

@@ -14,6 +14,7 @@ typer = "^0.12.5"
 fastapi = {extras = ["standard"], version = "^0.115.2"}
 azure-functions = "^1.21.3"
 pydantic-settings = "^2.5.2"
+azure-iot-device = "^2.14.0"
 
 [tool.poetry.group.dev.dependencies]
 pre-commit = "^4.0.1"
@@ -33,6 +34,8 @@ build-backend = "poetry.core.masonry.api"
 [tool.ruff]
 line-length = 120
 target-version = "py310"
+
+[tool.ruff.lint]
 select = ["E", "F", "I", "UP"]
 ignore = ["D203"]
 

tests/test_iot.py

@@ -0,0 +1,14 @@
+from logging import getLogger
+
+from tests.utilities import client
+
+logger = getLogger(__name__)
+
+
+def test_iot():
+    path_format = "/iot_hub/{0}"
+    response = client.get(
+        url=path_format.format("device_twin"),
+    )
+    assert response.status_code == 200
+    logger.info(f"response: {response.json()}")

workshop_azure_iot/core.py

@@ -1,13 +1,15 @@
 from fastapi import FastAPI
 
 from workshop_azure_iot.routers.core import router as core_router
+from workshop_azure_iot.routers.iot_hub import router as iot_hub_router
 
 app = FastAPI(
     docs_url="/",
 )
 
 for router in [
     core_router,
+    iot_hub_router,
     # Add routers here
 ]:
     app.include_router(router)

workshop_azure_iot/internals/iot_hub.py

@@ -0,0 +1,20 @@
+from logging import getLogger
+
+from azure.iot.device.aio import IoTHubDeviceClient
+
+from workshop_azure_iot.settings.iot_hub import Settings
+
+logger = getLogger(__name__)
+
+
+class Client:
+    def __init__(self, settings: Settings) -> None:
+        self.settings = settings
+
+    async def get_device_twin(self) -> dict:
+        client = IoTHubDeviceClient.create_from_connection_string(self.settings.iot_hub_device_connection_string)
+        # FIXME: to make it faster, connection should be established once and reused
+        await client.connect()
+        twin = await client.get_twin()
+        await client.shutdown()
+        return twin

workshop_azure_iot/routers/iot_hub.py

@@ -0,0 +1,26 @@
+from logging import getLogger
+
+from fastapi import APIRouter, status
+from fastapi.responses import JSONResponse
+
+from workshop_azure_iot.internals.iot_hub import Client
+from workshop_azure_iot.settings.iot_hub import Settings
+
+logger = getLogger(__name__)
+
+client = Client(
+    settings=Settings(),
+)
+
+router = APIRouter(
+    prefix="/iot_hub",
+    tags=["iot_hub"],
+)
+
+
+@router.get("/device_twin")
+async def get_device_twin():
+    return JSONResponse(
+        status_code=status.HTTP_200_OK,
+        content=await client.get_device_twin(),
+    )

workshop_azure_iot/settings/iot_hub.py

@@ -0,0 +1,6 @@
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    iot_hub_device_connection_string: str
+    model_config = SettingsConfigDict(env_file="iot_hub.env")