implementing the gl.edge pattern

Author: timmarkhuff

src/groundlight/__init__.py

@@ -7,7 +7,7 @@
 
 # Imports from our code
 from .client import Groundlight
-from .client import GroundlightClientError, ApiTokenError, NotFoundError
+from .client import GroundlightClientError, ApiTokenError, EdgeNotAvailableError, NotFoundError
 from .experimental_api import ExperimentalApi
 from .binary_labels import Label
 from .version import get_version

src/groundlight/client.py

@@ -69,6 +69,12 @@ class ApiTokenError(GroundlightClientError):
     pass
 
 
+class EdgeNotAvailableError(GroundlightClientError):
+    """Raised when an edge-only method is called against a non-edge endpoint."""
+
+    pass
+
+
 class Groundlight:  # pylint: disable=too-many-instance-attributes,too-many-public-methods
     """
     Client for accessing the Groundlight cloud service. Provides methods to create visual detectors,

src/groundlight/edge/__init__.py

@@ -1,3 +1,4 @@
+from .api import EdgeAPI
 from .config import (
     DEFAULT,
     DISABLED,
@@ -14,6 +15,7 @@
     "DEFAULT",
     "DISABLED",
     "EDGE_ANSWERS_WITH_ESCALATION",
+    "EdgeAPI",
     "NO_CLOUD",
     "DetectorsConfig",
     "DetectorConfig",

src/groundlight/edge/api.py

@@ -0,0 +1,78 @@
+import time
+
+import requests
+
+from groundlight.client import EdgeNotAvailableError
+from groundlight.edge.config import EdgeEndpointConfig
+
+
+_EDGE_METHOD_UNAVAILABLE_HINT = (
+    "Make sure the client is pointed at a running edge endpoint "
+    "(via GROUNDLIGHT_ENDPOINT env var or the endpoint= constructor arg)."
+)
+
+
+class EdgeAPI:
+    """Namespace for edge-endpoint operations, accessed via ``gl.edge``."""
+
+    def __init__(self, client) -> None:
+        self._client = client
+
+    def _base_url(self) -> str:
+        return self._client._edge_base_url()
+
+    def _request(self, method: str, path: str, **kwargs) -> requests.Response:
+        url = f"{self._base_url()}{path}"
+        headers = self._client.get_raw_headers()
+        try:
+            response = requests.request(
+                method, url, headers=headers, verify=self._client.configuration.verify_ssl, **kwargs
+            )
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as e:
+            if e.response is not None and e.response.status_code == 404:
+                raise EdgeNotAvailableError(f"Edge method not available at {url}. {_EDGE_METHOD_UNAVAILABLE_HINT}") from e
+            raise
+        except requests.exceptions.ConnectionError as e:
+            raise EdgeNotAvailableError(f"Could not connect to {self._base_url()}. {_EDGE_METHOD_UNAVAILABLE_HINT}") from e
+        return response
+
+    def get_config(self) -> EdgeEndpointConfig:
+        """Retrieve the active edge endpoint configuration."""
+        response = self._request("GET", "/edge-config")
+        return EdgeEndpointConfig.from_payload(response.json())
+
+    def get_detector_readiness(self) -> dict[str, bool]:
+        """Check which configured detectors have inference pods ready to serve.
+
+        :return: Dict mapping detector_id to readiness (True/False).
+        """
+        response = self._request("GET", "/edge-detector-readiness")
+        return {det_id: info["ready"] for det_id, info in response.json().items()}
+
+    def set_config(
+        self,
+        config: EdgeEndpointConfig,
+        timeout_sec: float = 600,
+    ) -> EdgeEndpointConfig:
+        """Replace the edge endpoint configuration and wait until all detectors are ready.
+
+        :param config: The new configuration to apply.
+        :param timeout_sec: Max seconds to wait for all detectors to become ready.
+        :return: The applied configuration as reported by the edge endpoint.
+        """
+        self._request("PUT", "/edge-config", json=config.to_payload())
+
+        poll_interval_seconds = 1
+        desired_ids = {d.detector_id for d in config.detectors if d.detector_id}
+        deadline = time.time() + timeout_sec
+        while time.time() < deadline:
+            readiness = self.get_detector_readiness()
+            if desired_ids and all(readiness.get(did, False) for did in desired_ids):
+                return self.get_config()
+            time.sleep(poll_interval_seconds)
+
+        raise TimeoutError(
+            f"Edge detectors were not all ready within {timeout_sec}s. "
+            "The edge endpoint may still be converging, or may have encountered an error."
+        )

src/groundlight/experimental_api.py

@@ -41,6 +41,7 @@
 )
 from urllib3.response import HTTPResponse
 
+from groundlight.edge.api import EdgeAPI
 from groundlight.edge.config import EdgeEndpointConfig
 from groundlight.images import parse_supported_image_types
 from groundlight.internalapi import _generate_request_id
@@ -105,6 +106,14 @@ def __init__(
         self.detector_reset_api = DetectorResetApi(self.api_client)
 
         self.edge_api = EdgeApi(self.api_client)
+        self._edge: EdgeAPI | None = None
+
+    @property
+    def edge(self) -> "EdgeAPI":
+        """Access edge-endpoint operations (e.g. ``gl.edge.get_config()``)."""
+        if self._edge is None:
+            self._edge = EdgeAPI(self)
+        return self._edge
 
     ITEMS_PER_PAGE = 100
 
@@ -828,66 +837,17 @@ def _edge_base_url(self) -> str:
         return urlunparse((parsed.scheme, parsed.netloc, "", "", "", ""))
 
     def get_edge_config(self) -> EdgeEndpointConfig:
-        """Retrieve the active edge endpoint configuration.
-
-        Only works when the client is pointed at an edge endpoint
-        (via GROUNDLIGHT_ENDPOINT or the endpoint constructor arg).
-        """
-        url = f"{self._edge_base_url()}/edge-config"
-        headers = self.get_raw_headers()
-        response = requests.get(url, headers=headers, verify=self.configuration.verify_ssl)
-        response.raise_for_status()
-        return EdgeEndpointConfig.from_payload(response.json())
+        """Deprecated: use ``gl.edge.get_config()`` instead."""
+        return self.edge.get_config()
 
     def get_edge_detector_readiness(self) -> dict[str, bool]:
-        """Check which configured detectors have inference pods ready to serve.
-
-        Only works when the client is pointed at an edge endpoint.
-
-        :return: Dict mapping detector_id to readiness (True/False).
-        """
-        url = f"{self._edge_base_url()}/edge-detector-readiness"
-        headers = self.get_raw_headers()
-        response = requests.get(url, headers=headers, verify=self.configuration.verify_ssl)
-        response.raise_for_status()
-        return {det_id: info["ready"] for det_id, info in response.json().items()}
+        """Deprecated: use ``gl.edge.get_detector_readiness()`` instead."""
+        return self.edge.get_detector_readiness()
 
     def set_edge_config(
         self,
         config: EdgeEndpointConfig,
-        mode: str = "REPLACE",
         timeout_sec: float = 300,
-        poll_interval_sec: float = 1,
     ) -> EdgeEndpointConfig:
-        """Send a new edge endpoint configuration and wait until all detectors are ready.
-
-        Only works when the client is pointed at an edge endpoint.
-
-        :param config: The new configuration to apply.
-        :param mode: Currently only "REPLACE" is supported.
-        :param timeout_sec: Max seconds to wait for all detectors to become ready.
-        :param poll_interval_sec: How often to poll readiness while waiting.
-        :return: The applied configuration as reported by the edge endpoint.
-        """
-        if mode != "REPLACE":
-            raise ValueError(f"Unsupported mode: {mode!r}. Currently only 'REPLACE' is supported.")
-
-        url = f"{self._edge_base_url()}/edge-config"
-        headers = self.get_raw_headers()
-        response = requests.put(
-            url, json=config.to_payload(), headers=headers, verify=self.configuration.verify_ssl
-        )
-        response.raise_for_status()
-
-        desired_ids = {d.detector_id for d in config.detectors if d.detector_id}
-        deadline = time.time() + timeout_sec
-        while time.time() < deadline:
-            readiness = self.get_edge_detector_readiness()
-            if desired_ids and all(readiness.get(did, False) for did in desired_ids):
-                return self.get_edge_config()
-            time.sleep(poll_interval_sec)
-
-        raise TimeoutError(
-            f"Edge detectors were not all ready within {timeout_sec}s. "
-            "The edge endpoint may still be converging."
-        )
+        """Deprecated: use ``gl.edge.set_config()`` instead."""
+        return self.edge.set_config(config, timeout_sec=timeout_sec)