fix(#695): add backend-neutral URI resolution

docs/changelog.rst

@@ -17,6 +17,10 @@ v0.59.0 - Data dictionary and loader access
 * :attr:`SQLSpec.loader <sqlspec.base.SQLSpec.loader>` gives read-only access to
   the registry's SQL file loader. SQLSpec creates the loader on first access
   when one was not supplied.
+* Object storage backends expose ``resolve_uri(path)`` to format one
+  backend-relative path as an absolute local path or protocol-qualified remote
+  URI without storage I/O. Custom backends that implement
+  ``ObjectStoreProtocol`` must define it too.
 
 **Changed:**
 

docs/reference/storage.rst

@@ -39,6 +39,26 @@ the regular Arrow read APIs for CSV, Arrow IPC, JSON, and JSONL payloads.
 Closing a sync generator or calling ``aclose()`` on its async iterator closes
 the active storage reader.
 
+Resolve portable storage addresses
+==================================
+
+Use ``backend.resolve_uri(path)`` when another component needs the address of
+an object. The method accepts the same backend-relative path as the read and
+write methods. It does not perform storage I/O, require the target to exist, or
+change the backend's configured path prefixes.
+
+For example, fsspec and obstore both resolve ``data.parquet`` against
+``s3://bucket/prefix`` with ``base_path="workspaces"`` as
+``s3://bucket/prefix/workspaces/data.parquet``. Local, fsspec-file, and
+obstore-file backends return the equivalent absolute filesystem path. Consumers
+should use this method instead of joining ``base_path``, ``base_uri``, or
+``store_uri`` themselves.
+
+The returned address is unsigned. Use ``sign_sync()`` or ``sign_async()`` when
+a caller needs a time-limited access grant. ``resolve_uri()`` does not sign,
+percent-encode, or validate the address, and its argument is an object key rather
+than an already-qualified remote URI.
+
 Pipelines
 =========
 

sqlspec/protocols.py

@@ -481,6 +481,10 @@ class ObjectStoreProtocol(Protocol):
     def __init__(self, uri: str, **kwargs: Any) -> None:
         return
 
+    def resolve_uri(self, path: "str | Path") -> str:
+        """Resolve a backend-relative path to its unsigned address."""
+        return ""
+
     def read_bytes_sync(self, path: "str | Path", **kwargs: Any) -> bytes:
         """Read bytes from an object synchronously."""
         return b""

sqlspec/storage/backends/base.py

@@ -12,6 +12,7 @@
 from typing_extensions import Self
 
 if TYPE_CHECKING:
+    from pathlib import Path
     from types import TracebackType
 
     from sqlspec.typing import ArrowRecordBatch, ArrowTable
@@ -167,6 +168,11 @@ class ObjectStoreBase:
 
     __slots__ = ()
 
+    @abstractmethod
+    def resolve_uri(self, path: "str | Path") -> str:
+        """Resolve a backend-relative path to its unsigned address."""
+        raise NotImplementedError
+
     @abstractmethod
     def read_bytes_sync(self, path: str, **kwargs: Any) -> bytes:
         """Read bytes from storage synchronously."""

sqlspec/storage/backends/fsspec.py

@@ -111,6 +111,21 @@ def from_config(cls, config: "dict[str, Any]") -> "FSSpecBackend":
     def base_uri(self) -> str:
         return self._fs_uri
 
+    def resolve_uri(self, path: str | Path) -> str:
+        """Resolve a backend-relative path to an unsigned address.
+
+        Args:
+            path: The same backend-relative path accepted by read and write methods.
+
+        Returns:
+            An absolute filesystem path for ``file`` or a protocol-qualified URI
+            for other filesystems. The target does not need to exist.
+        """
+        resolved_path = self._resolve_path(path)
+        if self.protocol == "file":
+            return str(Path(resolved_path).resolve())
+        return str(self.fs.unstrip_protocol(resolved_path))
+
     def _resolve_path(self, path: str | Path) -> str:
         return resolve_storage_path(path, self.base_path, self.protocol, strip_file_scheme=False)
 

sqlspec/storage/backends/local.py

@@ -76,6 +76,17 @@ def __init__(self, uri: str = "", **kwargs: Any) -> None:
 
         self.protocol = "file"
 
+    def resolve_uri(self, path: "str | Path") -> str:
+        """Resolve a backend-relative path to an absolute filesystem path.
+
+        Args:
+            path: The same backend-relative path accepted by read and write methods.
+
+        Returns:
+            The absolute filesystem path. The target does not need to exist.
+        """
+        return str(self._resolve_path(path).resolve())
+
     def _resolve_path(self, path: "str | Path") -> Path:
         """Resolve path relative to base_path.
 

sqlspec/storage/backends/obstore.py

@@ -199,6 +199,30 @@ def from_config(cls, config: "dict[str, Any]") -> "ObStoreBackend":
 
         return cls(uri=store_uri, **kwargs)
 
+    def resolve_uri(self, path: "str | Path") -> str:
+        """Resolve a backend-relative path to an unsigned address.
+
+        Args:
+            path: The same backend-relative path accepted by read and write methods.
+
+        Returns:
+            An absolute filesystem path for local stores or a protocol-qualified
+            URI for remote stores. The target does not need to exist.
+        """
+        resolved_path = self._resolve_path(path)
+        if self._is_local_store:
+            return str((Path(self._local_store_root) / resolved_path).resolve())
+
+        parsed = urlparse(self.store_uri)
+        joined_path = "/".join(part.strip("/") for part in (parsed.path, resolved_path) if part.strip("/"))
+        authority = f"{parsed.scheme}://{parsed.netloc}"
+        address = f"{authority}/{joined_path}" if parsed.netloc else f"{authority}{joined_path}"
+        if parsed.query:
+            address = f"{address}?{parsed.query}"
+        if parsed.fragment:
+            address = f"{address}#{parsed.fragment}"
+        return address
+
     def _resolve_path(self, path: "str | Path") -> str:
         if self._is_local_store:
             return self._local_store_path(path)

tests/integration/storage/test_integration.py

@@ -485,6 +485,23 @@ def test_backend_consistency(request: pytest.FixtureRequest, backend_name: str)
             backend.sign_sync(test_path, expires_in=3600)
 
 
+def test_s3_backend_resolved_uri_consistency(
+    fsspec_s3_backend_optional: "ObjectStoreProtocol",
+    obstore_s3_backend_optional: "ObjectStoreProtocol",
+    rustfs_bucket_name: str,
+) -> None:
+    """Fsspec and obstore expose the same URI after equivalent S3 operations."""
+    test_path = "resolved_uri_consistency.txt"
+
+    for backend in (fsspec_s3_backend_optional, obstore_s3_backend_optional):
+        backend.write_text_sync(test_path, TEST_TEXT_CONTENT)
+        assert backend.read_text_sync(test_path) == TEST_TEXT_CONTENT
+
+    expected = f"s3://{rustfs_bucket_name}/{test_path}"
+    assert fsspec_s3_backend_optional.resolve_uri(test_path) == expected
+    assert obstore_s3_backend_optional.resolve_uri(test_path) == expected
+
+
 @pytest.mark.parametrize("backend_name", ["local_backend", "fsspec_s3_backend_optional", "obstore_s3_backend_optional"])
 async def test_backend_async_consistency(request: pytest.FixtureRequest, backend_name: str) -> None:
     """Test that all backends provide consistent async behavior."""

tests/unit/storage/test_uri_resolution.py

@@ -0,0 +1,101 @@
+"""Unit tests for backend-neutral storage URI resolution."""
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from sqlspec.protocols import ObjectStoreProtocol
+from sqlspec.storage.backends.base import ObjectStoreBase
+from sqlspec.storage.backends.local import LocalStore
+from sqlspec.typing import FSSPEC_INSTALLED, OBSTORE_INSTALLED
+
+
+def test_object_store_contract_exposes_uri_resolution() -> None:
+    assert hasattr(ObjectStoreProtocol, "resolve_uri")
+    assert getattr(ObjectStoreBase.resolve_uri, "__isabstractmethod__", False)
+
+
+def test_local_store_resolves_nonexistent_path(tmp_path: Path) -> None:
+    store = LocalStore(str(tmp_path), base_path="workspaces")
+
+    assert store.resolve_uri(Path("data.parquet")) == str((tmp_path / "workspaces" / "data.parquet").resolve())
+
+
+@pytest.mark.skipif(not FSSPEC_INSTALLED, reason="fsspec missing")
+def test_fsspec_resolves_prefixed_s3_uri() -> None:
+    from sqlspec.storage.backends.fsspec import FSSpecBackend
+
+    store = FSSpecBackend("s3://bucket/prefix/", base_path="workspaces/")
+
+    assert store.resolve_uri(Path("data.parquet")) == "s3://bucket/prefix/workspaces/data.parquet"
+
+
+@pytest.mark.skipif(not FSSPEC_INSTALLED, reason="fsspec missing")
+def test_fsspec_protocol_only_config_resolves_bucket_prefix() -> None:
+    from sqlspec.storage.backends.fsspec import FSSpecBackend
+
+    store = FSSpecBackend.from_config({"protocol": "s3", "base_path": "bucket/prefix", "fs_config": {}})
+
+    assert store.resolve_uri("data.parquet") == "s3://bucket/prefix/data.parquet"
+
+
+@pytest.mark.skipif(not FSSPEC_INSTALLED, reason="fsspec missing")
+def test_fsspec_resolves_memory_uri_with_empty_base_path() -> None:
+    from sqlspec.storage.backends.fsspec import FSSpecBackend
+
+    store = FSSpecBackend("memory://")
+
+    assert store.resolve_uri("data.parquet") == "memory://data.parquet"
+
+
+@pytest.mark.skipif(not OBSTORE_INSTALLED, reason="obstore missing")
+def test_obstore_resolves_prefixed_s3_uri() -> None:
+    store = _remote_obstore("s3://bucket/prefix/", base_path="workspaces/")
+
+    assert store.resolve_uri(Path("data.parquet")) == "s3://bucket/prefix/workspaces/data.parquet"
+
+
+@pytest.mark.skipif(not OBSTORE_INSTALLED, reason="obstore missing")
+def test_obstore_preserves_query_and_fragment() -> None:
+    store = _remote_obstore("s3://bucket/prefix?version=1#section", base_path="workspaces")
+
+    assert store.resolve_uri("data.parquet") == "s3://bucket/prefix/workspaces/data.parquet?version=1#section"
+
+
+@pytest.mark.skipif(not OBSTORE_INSTALLED, reason="obstore missing")
+def test_obstore_resolves_memory_uri_without_corrupting_empty_authority() -> None:
+    from sqlspec.storage.backends.obstore import ObStoreBackend
+
+    store = ObStoreBackend("memory://")
+
+    assert store.resolve_uri("data.parquet") == "memory://data.parquet"
+
+
+@pytest.mark.skipif(not FSSPEC_INSTALLED or not OBSTORE_INSTALLED, reason="storage backends missing")
+def test_file_backends_resolve_equivalent_absolute_paths(tmp_path: Path) -> None:
+    from sqlspec.storage.backends.fsspec import FSSpecBackend
+    from sqlspec.storage.backends.obstore import ObStoreBackend
+
+    expected = str((tmp_path / "workspaces" / "data.parquet").resolve())
+    backends = (
+        LocalStore(str(tmp_path), base_path="workspaces"),
+        FSSpecBackend(f"file://{tmp_path}", base_path="workspaces"),
+        ObStoreBackend(f"file://{tmp_path}", base_path="workspaces"),
+    )
+
+    assert [backend.resolve_uri("data.parquet") for backend in backends] == [expected, expected, expected]
+
+
+def _remote_obstore(store_uri: str, base_path: str = "") -> Any:
+    from sqlspec.storage.backends.obstore import ObStoreBackend
+
+    store = ObStoreBackend.__new__(ObStoreBackend)
+    store._is_local_store = False
+    store._local_store_root = ""
+    store.base_path = base_path.rstrip("/")
+    store.protocol = store_uri.split("://", maxsplit=1)[0]
+    store.store = object()
+    store.store_options = {}
+    store.store_uri = store_uri
+    return store