✨ Extend UPath.to_url() to private paths

lamindb_setup/core/upath.py

@@ -531,8 +531,8 @@ def synchronize_to(
         # no need to cast local_stat.st_mtime to int
         # because if it has the fractional part and cloud_mtime doesn't
         # and they have the same integer part then cloud_mtime can't be bigger
-        is_sync_needed = (
-            lambda cloud_mtime, local_stat: cloud_mtime > local_stat.st_mtime
+        is_sync_needed = lambda cloud_mtime, local_stat: (
+            cloud_mtime > local_stat.st_mtime
         )
 
     local_paths: list[Path] = []
@@ -808,27 +808,56 @@ def view_tree(
     logger.print(message)
 
 
-def to_url(upath: S3Path) -> str:
-    """Public storage URL.
+def to_url(upath: UPath) -> str:
+    """Generates a URL for an object represented by `UPath`.
 
-    Generates a public URL for an object in an S3 bucket using fsspec's UPath,
-    considering the bucket's region.
+    For S3/GCS paths, this returns a public URL considering the bucket region.
+    If the S3 path is not publicly hosted, it returns a LaminHub URL if the artifact is hosted on LaminHub.
 
     Args:
-        upath: A `UPath` object representing an S3 path.
+        upath: A `UPath` object.
 
     Returns:
-        A string containing the public URL to the S3 object.
+        A string containing the URL to the object.
     """
-    if upath.protocol != "s3":
-        raise ValueError("The provided UPath must be an S3 path.")
-    key = "/".join(upath.parts[1:])
-    bucket = upath.drive
-    region = get_storage_region(upath)
-    if region == "us-east-1":
-        return f"https://{bucket}.s3.amazonaws.com/{key}"
-    else:
-        return f"https://{bucket}.s3-{region}.amazonaws.com/{key}"
+    from ._settings import settings
+
+    if upath.protocol == "s3":
+        key = "/".join(upath.parts[1:])
+        bucket = upath.drive
+        if _is_publicly_accessible_path(upath):
+            region = get_storage_region(upath)
+            if region == "us-east-1":
+                return f"https://{bucket}.s3.amazonaws.com/{key}"
+            return f"https://{bucket}.s3-{region}.amazonaws.com/{key}"
+        elif settings.instance.is_on_hub:
+            origin = settings.instance.ui_url
+            if origin is not None:
+                common = f"{origin}/storage/s3/{bucket}%2F/{key}"
+                return common
+        else:
+            raise ValueError(
+                "The provided S3 UPath must be publicly accessible or the artifact must be hosted on LaminHub."
+            )
+    if upath.protocol == "gs":
+        if _is_publicly_accessible_path(upath):
+            return f"https://storage.googleapis.com/{str(upath).removeprefix('gs://')}"
+        else:
+            raise ValueError(
+                "This function only supports publicly accessible GCS paths."
+            )
+    if upath.protocol in {"http", "https"}:
+        return str(upath)
+    raise ValueError("The provided UPath must be an S3, GCS, HTTP, or HTTPS path.")
+
+
+def _is_publicly_accessible_path(upath: UPath) -> bool:
+    """Check whether an S3/GCS path is anonymously readable."""
+    anon_path = UPath(upath, anon=True)
+    try:
+        return anon_path.exists()
+    except Exception:
+        return False
 
 
 def from_auth(cls, path: AnyPathStr) -> UPath:

tests/core/test_to_url.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+from types import SimpleNamespace
+
 import lamindb_setup as ln_setup
+from lamindb_setup.core._settings import settings
 
 
 def test_to_url():
@@ -30,3 +33,50 @@ def test_to_url():
         ).to_url()
         == "https://lamin-eu-central-1.s3-eu-central-1.amazonaws.com/9fm7UN13/test-folder"
     )
+
+
+def test_to_url_gcs_root():
+    upath = ln_setup.core.upath.UPath(
+        "gs://rxrx1-europe-west4/images/test/HEPG2-08/Plate1/B02_s1_w1.png"
+    )
+    assert (
+        upath.to_url()
+        == "https://storage.googleapis.com/rxrx1-europe-west4/images/test/HEPG2-08/Plate1/B02_s1_w1.png"
+    )
+
+
+def test_to_url_https_root():
+    upath = ln_setup.core.upath.UPath("https://example.com/files/document.txt")
+    assert upath.to_url() == "https://example.com/files/document.txt"
+
+
+def test_to_url_s3_hub_private_route(monkeypatch):
+    monkeypatch.setattr(ln_setup.core.upath, "_is_public_s3_path", lambda _: False)
+    monkeypatch.setattr(
+        settings,
+        "_instance_settings",
+        SimpleNamespace(is_on_hub=True, ui_url="https://app.lamin.ai"),
+        raising=False,
+    )
+    upath = ln_setup.core.upath.UPath("s3://lamindb-ci/test-data/test.parquet")
+    assert (
+        upath.to_url()
+        == "https://app.lamin.ai/storage/s3/lamindb-ci%2F/test-data/test.parquet"
+    )
+
+
+def test_to_url_s3_public_stays_native(monkeypatch):
+    monkeypatch.setattr(ln_setup.core.upath, "_is_public_s3_path", lambda _: True)
+    monkeypatch.setattr(
+        ln_setup.core.upath, "get_storage_region", lambda _: "us-east-1"
+    )
+    monkeypatch.setattr(
+        settings,
+        "_instance_settings",
+        SimpleNamespace(is_on_hub=True, ui_url="https://app.lamin.ai"),
+        raising=False,
+    )
+    upath = ln_setup.core.upath.UPath("s3://lamindb-ci/test-data/test.parquet")
+    assert (
+        upath.to_url() == "https://lamindb-ci.s3.amazonaws.com/test-data/test.parquet"
+    )