py: add basic service URL resolver

Signed-off-by: Isabella do Amaral <idoamara@redhat.com>

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.26.0", optional = true }
+kubernetes = "^31.0.0"
 
 [tool.poetry.extras]
 hf = ["huggingface-hub"]

clients/python/src/model_registry/_client.py

@@ -22,6 +22,11 @@
 ModelTypes = t.Union[RegisteredModel, ModelVersion, ModelArtifact]
 TModel = t.TypeVar("TModel", bound=ModelTypes)
 
+DSC_CRD = "datasciencecluster.opendatahub.io/v1"
+DEFAULT_NS = "kubeflow"
+DSC_NS_CONFIG = "registriesNamespace"
+EXTERNAL_ADDR_ANNOTATION = "routing.opendatahub.io/external-address-rest"
+
 
 class ModelRegistry:
     """Model registry client."""
@@ -87,6 +92,76 @@ 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()
+            g, v = DSC_CRD.split("/")
+            p = f"{g.split('.')[0]}s"
+            try:
+                dsc_raw = kcustom.list_cluster_custom_object(
+                    group=g,
+                    version=v,
+                    plural=p,
+                )
+            except client.ApiException as e:
+                msg = f"Failed to list {p}: {e}"
+                warn(msg, stacklevel=2)
+                ns = DEFAULT_NS
+            else:
+                ns = t.cast(
+                    dict[str, t.Any],
+                    dsc_raw["items"][0],
+                )["status"]["components"]["modelregistry"][DSC_NS_CONFIG]
+
+        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(
+            EXTERNAL_ADDR_ANNOTATION
+        )
+        if ext_addr:
+            host, port = ext_addr.split(":")
+            host = f"https://{host}"
+            port = int(port)
+        elif not is_secure:
+            host = f"http://{meta.name}"
+            port = next(
+                (
+                    int(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(
+            host,
+            port,
+            author=author,
+        )
+
     def async_runner(self, coro: t.Any) -> t.Any:
         import asyncio