py: add basic service URL resolver

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

clients/python/README.md

@@ -37,18 +37,18 @@ pip install huggingface-hub
 
 ### Connecting to MR
 
-You can connect to a secure Model Registry using the default constructor (recommended):
+You can connect to a secure Model Registry using the service constructor (recommended):
 
 ```py
 from model_registry import ModelRegistry
 
-registry = ModelRegistry("https://server-address", author="Ada Lovelace")  # Defaults to a secure connection via port 443
+registry = ModelRegistry.from_service("modelregistry-sample", "Ada Lovelace")  # Defaults to a secure connection via port 443
 ```
 
 Or you can set the `is_secure` flag to `False` to connect **without** TLS (not recommended):
 
 ```py
-registry = ModelRegistry("http://server-address", 8080, author="Ada Lovelace", is_secure=False)  # insecure port set to 8080
+registry = ModelRegistry.from_service("modelregistry-sample", "Ada Lovelace", is_secure=False)  # insecure port set to 8080
 ```
 
 ### Registering models
@@ -190,6 +190,14 @@ This is necessary as the test suite will manage a Model Registry server and an M
 each run.
 You can use `make test` to execute `pytest`.
 
+### Connecting to MR outside a cluster
+
+You can simply use the default `ModelRegistry` constructor:
+
+```py
+registry = ModelRegistry("http://server-address", 8080, author="Ada Lovelace", is_secure=False)  # insecure port set to 8080
+```
+
 ### Running Locally on Mac M1 or M2 (arm64 architecture)
 
 Check out our [recommendations on setting up your docker engine](https://github.com/kubeflow/model-registry/blob/main/CONTRIBUTING.md#docker-engine) on an ARM processor.

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