Add reusable identity provisioning adapter

src/user_engine/adapters/provisioning.py

@@ -0,0 +1,67 @@
+"""HTTP adapter for the NetKingdom identity provisioning service."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+from urllib.error import HTTPError
+from urllib.request import Request, urlopen
+
+from user_engine.ports import ProvisioningRequest, ProvisioningResult
+
+
+class HTTPIdentityProvisioningAdapter:
+    def __init__(self, *, base_url: str, bearer_token: str, timeout: float = 10) -> None:
+        self.base_url = base_url.rstrip("/")
+        self.bearer_token = bearer_token
+        self.timeout = timeout
+
+    def provision(self, request: ProvisioningRequest) -> ProvisioningResult:
+        return self._post("/v1/identities/provision", {
+            "user_id": request.user_id,
+            "tenant": request.tenant,
+            "primary_email": request.primary_email,
+            "display_name": request.display_name,
+            "idempotency_key": request.idempotency_key,
+            "correlation_id": request.correlation_id,
+            "roles": request.roles,
+        })
+
+    def suspend(self, *, external_subject: str, idempotency_key: str, correlation_id: str) -> ProvisioningResult:
+        return self._lifecycle("suspend", external_subject, idempotency_key, correlation_id)
+
+    def reactivate(self, *, external_subject: str, idempotency_key: str, correlation_id: str) -> ProvisioningResult:
+        return self._lifecycle("reactivate", external_subject, idempotency_key, correlation_id)
+
+    def deprovision(self, *, external_subject: str, idempotency_key: str, correlation_id: str) -> ProvisioningResult:
+        return self._lifecycle("deprovision", external_subject, idempotency_key, correlation_id)
+
+    def _lifecycle(self, action: str, subject: str, key: str, correlation_id: str) -> ProvisioningResult:
+        return self._post(f"/v1/identities/{action}", {
+            "external_subject": subject,
+            "idempotency_key": key,
+            "correlation_id": correlation_id,
+        })
+
+    def _post(self, path: str, payload: dict[str, Any]) -> ProvisioningResult:
+        request = Request(
+            self.base_url + path,
+            data=json.dumps(payload).encode(),
+            headers={
+                "Authorization": f"Bearer {self.bearer_token}",
+                "Content-Type": "application/json",
+            },
+            method="POST",
+        )
+        try:
+            with urlopen(request, timeout=self.timeout) as response:
+                result = json.loads(response.read())
+        except HTTPError as exc:
+            message = exc.read(4096).decode("utf-8", "replace")
+            raise RuntimeError(f"identity provisioning failed ({exc.code}): {message}") from exc
+        return ProvisioningResult(
+            provider=str(result["provider"]),
+            external_subject=str(result["external_subject"]),
+            status=str(result["status"]),
+            resumed=bool(result.get("resumed", False)),
+        )