fix(signed url): pass version id down to fsspec via path

src/datachain/catalog/catalog.py

@@ -1236,10 +1236,16 @@ def ls_dataset_rows(
 
         return q.to_db_records()
 
-    def signed_url(self, source: str, path: str, client_config=None) -> str:
+    def signed_url(
+        self,
+        source: str,
+        path: str,
+        version_id: Optional[str] = None,
+        client_config=None,
+    ) -> str:
         client_config = client_config or self.client_config
         client = Client.get_client(source, self.cache, **client_config)
-        return client.url(path)
+        return client.url(path, version_id=version_id)
 
     def export_dataset_table(
         self,

src/datachain/client/azure.py

@@ -1,4 +1,5 @@
-from typing import Any
+from typing import Any, Optional
+from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
 
 from adlfs import AzureBlobFileSystem
 from tqdm import tqdm
@@ -57,4 +58,25 @@ async def _fetch_flat(self, start_prefix: str, result_queue: ResultQueue) -> Non
         finally:
             result_queue.put_nowait(None)
 
+    @classmethod
+    def _split_version(cls, path: str) -> tuple[str, Optional[str]]:
+        parts = list(urlsplit(path))
+        query = parse_qs(parts[3])
+        if "versionid" in query:
+            version_id = query["versionid"][0]
+            del query["versionid"]
+            parts[3] = urlencode(query)
+        else:
+            version_id = None
+        return urlunsplit(parts), version_id
+
+    @classmethod
+    def _join_version(cls, path: str, version_id: Optional[str]) -> str:
+        parts = list(urlsplit(path))
+        query = parse_qs(parts[3])
+        if "versionid" in query:
+            raise ValueError("path already includes a version query")
+        parts[3] = f"versionid={version_id}" if version_id else ""
+        return urlunsplit(parts)
+
     _fetch_default = _fetch_flat

src/datachain/client/fsspec.py

@@ -138,9 +138,18 @@ def create_fs(cls, **kwargs) -> "AbstractFileSystem":
         return fs
 
     @classmethod
-    def version_path(cls, path: str, version_id: Optional[str]) -> str:
+    def _split_version(cls, path: str) -> tuple[str, Optional[str]]:
+        return path, None
+
+    @classmethod
+    def _join_version(cls, path: str, version_id: Optional[str]) -> str:
         return path
 
+    @classmethod
+    def version_path(cls, path: str, version_id: Optional[str]) -> str:
+        path, _ = cls._split_version(path)
+        return cls._join_version(path, version_id)
+
     @classmethod
     def from_name(
         cls,
@@ -202,7 +211,11 @@ def fs(self) -> "AbstractFileSystem":
         return self._fs
 
     def url(self, path: str, expires: int = 3600, **kwargs) -> str:
-        return self.fs.sign(self.get_full_path(path), expiration=expires, **kwargs)
+        return self.fs.sign(
+            self.get_full_path(path, kwargs.pop("version_id", None)),
+            expiration=expires,
+            **kwargs,
+        )
 
     async def get_current_etag(self, file: "File") -> str:
         kwargs = {}

src/datachain/client/gcs.py

@@ -39,9 +39,13 @@ def url(self, path: str, expires: int = 3600, **kwargs) -> str:
         If the client is anonymous, a public URL is returned instead
         (see https://cloud.google.com/storage/docs/access-public-data#api-link).
         """
+        version_id = kwargs.pop("version_id", None)
         if self.fs.storage_options.get("token") == "anon":
-            return f"https://storage.googleapis.com/{self.name}/{path}"
-        return self.fs.sign(self.get_full_path(path), expiration=expires, **kwargs)
+            query = f"?generation={version_id}" if version_id else ""
+            return f"https://storage.googleapis.com/{self.name}/{path}{query}"
+        return self.fs.sign(
+            self.get_full_path(path, version_id), expiration=expires, **kwargs
+        )
 
     @staticmethod
     def parse_timestamp(timestamp: str) -> datetime:
@@ -150,8 +154,3 @@ def _join_version(cls, path: str, version_id: Optional[str]) -> str:
         if path_version:
             raise ValueError("path already includes an object generation")
         return f"{path}#{version_id}" if version_id else path
-
-    @classmethod
-    def version_path(cls, path: str, version_id: Optional[str]) -> str:
-        path, _ = cls._split_version(path)
-        return cls._join_version(path, version_id)

src/datachain/client/s3.py

@@ -1,5 +1,6 @@
 import asyncio
 from typing import Any, Optional, cast
+from urllib.parse import parse_qs, urlencode, urlsplit, urlunsplit
 
 from botocore.exceptions import NoCredentialsError
 from s3fs import S3FileSystem
@@ -121,6 +122,27 @@ def _entry_from_boto(self, v, bucket, versions=False) -> File:
             size=v["Size"],
         )
 
+    @classmethod
+    def _split_version(cls, path: str) -> tuple[str, Optional[str]]:
+        parts = list(urlsplit(path))
+        query = parse_qs(parts[3])
+        if "versionId" in query:
+            version_id = query["versionId"][0]
+            del query["versionId"]
+            parts[3] = urlencode(query)
+        else:
+            version_id = None
+        return urlunsplit(parts), version_id
+
+    @classmethod
+    def _join_version(cls, path: str, version_id: Optional[str]) -> str:
+        parts = list(urlsplit(path))
+        query = parse_qs(parts[3])
+        if "versionId" in query:
+            raise ValueError("path already includes a version query")
+        parts[3] = f"versionId={version_id}" if version_id else ""
+        return urlunsplit(parts)
+
     async def _fetch_dir(
         self,
         prefix,

tests/unit/test_client_gcs.py

@@ -4,3 +4,11 @@
 def test_anon_url():
     client = Client.get_client("gs://foo", None, anon=True)
     assert client.url("bar") == "https://storage.googleapis.com/foo/bar"
+
+
+def test_anon_versioned_url():
+    client = Client.get_client("gs://foo", None, anon=True)
+    assert (
+        client.url("bar", version_id="1234566")
+        == "https://storage.googleapis.com/foo/bar?generation=1234566"
+    )