fix(storage): resolve Windows SQL file paths

docs/changelog.rst

@@ -38,6 +38,8 @@ v0.59.0 - Data dictionary and loader access
 
 **Fixed:**
 
+* SQL files supplied with Windows drive paths now resolve from their requested
+  directory instead of the process working directory.
 * Async statement errors from mypyc-compiled drivers are translated into
   :class:`~sqlspec.exceptions.SQLSpecError` instead of terminating the process.
 * ADBC adapters for PostgreSQL now keep ``None`` in arrays. Each value binds as

sqlspec/storage/registry.py

@@ -9,7 +9,7 @@
 import re
 from pathlib import Path
 from typing import Any, Final, cast
-from urllib.parse import unquote, urlparse
+from urllib.parse import unquote, urlparse, urlunparse
 
 from mypy_extensions import mypyc_attr
 
@@ -126,11 +126,11 @@ def get(self, uri_or_alias: str | Path, *, backend: str | None = None, **kwargs:
                 file_path = strip_windows_drive_prefix(unquote(parsed.path))
 
                 path_obj = Path(file_path).expanduser().resolve()
-                base_uri = f"file://{path_obj.parent}" if is_file_destination(path_obj) else f"file://{path_obj}"
+                base_uri = _local_backend_uri(path_obj)
         elif is_local_path(path_str):
             scheme = "file"
             path_obj = Path(path_str).expanduser().resolve()
-            base_uri = f"file://{path_obj.parent}" if is_file_destination(path_obj) else f"file://{path_obj}"
+            base_uri = _local_backend_uri(path_obj)
         else:
             msg = f"Unknown storage alias or invalid URI: '{uri_or_alias}'"
             raise ImproperConfigurationError(msg)
@@ -314,3 +314,11 @@ def clear_aliases(self) -> None:
 
 
 storage_registry = StorageRegistry()
+
+
+def _local_backend_uri(path: Path) -> str:
+    root_path = path.parent if is_file_destination(path) else path
+    uri_path = root_path.as_posix()
+    if root_path.drive and not uri_path.startswith("/"):
+        uri_path = f"/{uri_path}"
+    return urlunparse(("file", "", uri_path, "", "", ""))

tests/unit/storage/test_registry_file_resolution.py

@@ -5,10 +5,12 @@
 """
 
 import tempfile
-from pathlib import Path
+from pathlib import Path, PureWindowsPath
+from typing import cast
 
 from sqlspec import SQLSpec
 from sqlspec.storage.backends.obstore import ObStoreBackend
+from sqlspec.storage.registry import _local_backend_uri
 
 
 def test_storage_registry_file_path_resolution_load_single_file_by_str_path() -> None:
@@ -26,7 +28,8 @@ def test_storage_registry_file_path_resolution_load_single_file_by_str_path() ->
 def test_storage_registry_file_path_resolution_load_single_file_by_path_object() -> None:
     """Loading a single SQL file by Path object should work."""
     with tempfile.TemporaryDirectory() as tmpdir:
-        sql_file = Path(tmpdir) / "hello.sql"
+        sql_file = Path(tmpdir) / "nested" / "sql" / "hello.sql"
+        sql_file.parent.mkdir(parents=True)
         sql_file.write_text("-- name: hello_world\nSELECT 1;\n")
         s = SQLSpec()
         s.load_sql_files(sql_file)
@@ -35,6 +38,13 @@ def test_storage_registry_file_path_resolution_load_single_file_by_path_object()
         assert "SELECT 1" in str(result)
 
 
+def test_storage_registry_file_path_resolution_builds_windows_file_uri() -> None:
+    """Windows drive paths should produce file URIs with an empty authority."""
+    sql_file = PureWindowsPath("C:/project/queries/nested/hello.sql")
+
+    assert _local_backend_uri(cast("Path", sql_file)) == "file:///C:/project/queries/nested"
+
+
 def test_storage_registry_file_path_resolution_load_from_directory() -> None:
     """Loading SQL files from a directory should work."""
     with tempfile.TemporaryDirectory() as tmpdir:

tests/unit/utils/test_mypyc_smoke.py

@@ -148,6 +148,8 @@ def test_construction_checks_build_provider_signatures_without_requiring_compila
         "statement_sentinel_identity",
         "sqlspec_construction",
     }
+    sqlspec_result = next(result for result in results if result["name"] == "sqlspec_construction")
+    assert sqlspec_result["error"] is None
 
 
 def test_statement_construction_checks_pass_without_requiring_compilation() -> None:

tools/scripts/mypyc_smoke.py

@@ -5,7 +5,9 @@
 import json
 import subprocess
 import sys
+import tempfile
 from collections.abc import Sequence
+from pathlib import Path
 from typing import Any, NamedTuple
 
 __all__ = ("SMOKE_IMPORTS", "SmokeImport", "is_compiled_module", "main", "run_construction_checks", "run_smoke")
@@ -106,6 +108,12 @@ def _check_sqlspec_construction() -> dict[str, Any]:
     result["compiled"] = is_compiled_module(base_module)
     try:
         manager = sqlspec_cls(loader=sql_file_loader_cls())
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sql_file = Path(tmpdir) / "nested" / "sql" / "smoke.sql"
+            sql_file.parent.mkdir(parents=True)
+            sql_file.write_text("-- name: mypyc_smoke_query\nSELECT 1;\n")
+            manager.load_sql_files(sql_file)
+            manager.get_sql("mypyc_smoke_query")
         config = manager.add_config(sqlite_config_cls(connection_config={"database": ":memory:"}))
         manager.event_channel(config)
         manager.telemetry_snapshot()