py: add basic service URL resolver

Signed-off-by: Isabella do Amaral <[email protected]>

clients/python/pyproject.toml

@@ -27,6 +27,7 @@ nest-asyncio = "^1.6.0"
 eval-type-backport = "^0.2.0"
 
 huggingface-hub = { version = ">=0.20.1,<0.25.0", optional = true }
+kubernetes = "^31.0.0"
 
 [tool.poetry.extras]
 hf = ["huggingface-hub"]

clients/python/src/model_registry/_client.py

@@ -87,6 +87,73 @@ def __init__(
                 server_address, port, user_token
             )
 
+    @classmethod
+    def from_service(
+        cls, name: str, author: str, *, ns: str | None = None, is_secure: bool = True
+    ) -> ModelRegistry:
+        """Create a client from a service name.
+
+        Args:
+            name: Service name.
+            author: Name of the author.
+
+        Keyword Args:
+            ns: Namespace. Defaults to DSC registriesNamespace, or `kubeflow` if unavailable.
+            is_secure: Whether to use a secure connection. Defaults to True.
+        """
+        from kubernetes import client, config
+
+        config.load_incluster_config()
+        if not ns:
+            kcustom = client.CustomObjectsApi()
+            try:
+                dsc_raw = kcustom.list_cluster_custom_object(
+                    group="datasciencecluster.opendatahub.io",
+                    version="v1",
+                    plural="datascienceclusters",
+                )
+            except client.ApiException as e:
+                msg = f"Failed to list datascienceclusters: {e}"
+                warn(msg, stacklevel=2)
+                # default
+                ns = "kubeflow"
+            else:
+                ns = t.cast(
+                    dict[str, t.Any],
+                    dsc_raw["items"][0],
+                )["status"]["components"]["modelregistry"]["registriesNamespace"]
+
+        kcore = client.CoreV1Api()
+        serv = t.cast(client.V1Service, kcore.read_namespaced_service(name, ns))
+        meta = t.cast(client.V1ObjectMeta, serv.metadata)
+        ext_addr = t.cast(dict[str, str], meta.annotations).get(
+            "routing.opendatahub.io/external-address-rest"
+        )
+        if ext_addr:
+            host, port = ext_addr.split(":")
+        elif not is_secure:
+            host = meta.name
+            port = next(
+                (
+                    str(port.port)
+                    for port in t.cast(
+                        list[client.V1ServicePort],
+                        t.cast(client.V1ServiceSpec, serv.spec).ports,
+                    )
+                    if port.app_protocol == "http"
+                ),
+                "8080",
+            )
+        else:
+            msg = "No external address found for secure connection"
+            raise StoreError(msg)
+
+        return cls(
+            f"https://{host}",
+            int(port),
+            author=author,
+        )
+
     def async_runner(self, coro: t.Any) -> t.Any:
         import asyncio