Merge pull request #1206 from transformerlab/fix/async-context-fsspec

Fix fsspec from inferring async context

Author: deep1401

lab-sdk/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "transformerlab"
-version = "0.0.64"
+version = "0.0.65"
 description = "Python SDK for Transformer Lab"
 readme = "README.md"
 requires-python = ">=3.10"

lab-sdk/src/lab/storage.py

@@ -129,35 +129,65 @@ def _get_storage_options() -> dict:
         return {}
 
 
-def _get_fs_and_root():
+def _get_fs_for_uri(uri: str):
     """
-    Initialize filesystem and root path from context variable or TFL_STORAGE_URI.
-    Falls back to local ~/.transformerlab or TFL_HOME_DIR when not set.
-    """
-    # Check context variable first, then fall back to environment variable
-    tfl_uri = _current_tfl_storage_uri.get() or os.getenv("TFL_STORAGE_URI")
+    Create a sync filesystem for a given URI.
+    Returns (filesystem, root_path) tuple.
 
-    if not tfl_uri or tfl_uri.strip() == "":
+    This explicitly creates a sync filesystem to avoid issues with async s3fs
+    passing storage_options to AioSession incorrectly when an event loop is running.
+    """
+    if not uri or uri.strip() == "":
         root = os.getenv(
             "TFL_HOME_DIR",
             os.path.join(os.path.expanduser("~"), ".transformerlab"),
         )
-        fs = fsspec.filesystem("file")
+        fs = fsspec.filesystem("file", asynchronous=False)
         return fs, root
 
-    # Get storage options based on cloud provider
-    storage_options = _get_storage_options()
-
-    # Let fsspec parse the URI
-    fs, _token, paths = fsspec.get_fs_token_paths(tfl_uri, storage_options=storage_options)
-    # For S3 and other remote filesystems, we need to maintain the full URI format
-    if tfl_uri.startswith(("s3://", "gs://", "abfs://", "gcs://")):
-        root = tfl_uri.rstrip("/")
+    # Extract protocol
+    if uri.startswith("s3://"):
+        protocol = "s3"
+    elif uri.startswith("gs://") or uri.startswith("gcs://"):
+        protocol = "gcs"
+    elif uri.startswith("abfs://"):
+        protocol = "abfs"
+    else:
+        # Local filesystem or unknown protocol - use fsspec's default handling
+        fs = fsspec.filesystem("file", asynchronous=False)
+        return fs, uri
+
+    # Build storage options as kwargs (not as nested dict)
+    # This ensures they're passed correctly to the filesystem, not to AioSession
+    fs_kwargs = {"asynchronous": False}  # Explicitly force sync mode
+    if protocol == "s3" and _AWS_PROFILE:
+        fs_kwargs["profile"] = _AWS_PROFILE
+    elif protocol in ("gcs", "gs") and _GCP_PROJECT:
+        fs_kwargs["project"] = _GCP_PROJECT
+
+    # Explicitly create sync filesystem to avoid async s3fs issues
+    # Use filesystem() directly with asynchronous=False to force sync version
+    # This prevents fsspec from auto-detecting async mode and creating async filesystem
+    fs = fsspec.filesystem(protocol, **fs_kwargs)
+
+    # For remote filesystems, maintain the full URI format as root
+    if uri.startswith(("s3://", "gs://", "abfs://", "gcs://")):
+        root = uri.rstrip("/")
     else:
-        root = paths[0] if paths else ""
+        root = uri
     return fs, root
 
 
+def _get_fs_and_root():
+    """
+    Initialize filesystem and root path from context variable or TFL_STORAGE_URI.
+    Falls back to local ~/.transformerlab or TFL_HOME_DIR when not set.
+    """
+    # Check context variable first, then fall back to environment variable
+    tfl_uri = _current_tfl_storage_uri.get() or os.getenv("TFL_STORAGE_URI")
+    return _get_fs_for_uri(tfl_uri)
+
+
 async def root_uri() -> str:
     _, root = _get_fs_and_root()
     return root
@@ -372,6 +402,7 @@ async def _get_uncached_filesystem(path: str, fs=None):
             if protocol:
                 # Create a new uncached filesystem with the same protocol and options
                 fs_kwargs = {
+                    "asynchronous": False,  # Explicitly force sync mode
                     "skip_instance_cache": True,
                     "default_fill_cache": False,
                     "use_listings_cache": False,
@@ -394,6 +425,7 @@ async def _get_uncached_filesystem(path: str, fs=None):
         # Create a new filesystem instance with caching disabled
         fs_uncached = fsspec.filesystem(
             protocol,
+            asynchronous=False,  # Explicitly force sync mode
             skip_instance_cache=True,
             default_fill_cache=False,
             use_listings_cache=False,
@@ -409,6 +441,7 @@ async def _get_uncached_filesystem(path: str, fs=None):
             storage_options = _get_storage_options()
             fs_uncached = fsspec.filesystem(
                 protocol,
+                asynchronous=False,  # Explicitly force sync mode
                 skip_instance_cache=True,
                 default_fill_cache=False,
                 use_listings_cache=False,
@@ -424,11 +457,11 @@ def _get_fs_for_path(path: str):
     """
     Get filesystem for a given path, handling S3 storage_options correctly.
     Returns (filesystem, parsed_path) tuple.
+
+    Reuses _get_fs_for_uri() to avoid code duplication.
     """
-    storage_options = {}
-    if path.startswith("s3://") and _AWS_PROFILE:
-        storage_options["profile"] = _AWS_PROFILE
-    return fsspec.core.url_to_fs(path, storage_options=storage_options if storage_options else None)
+    fs, _ = _get_fs_for_uri(path)
+    return fs, path
 
 
 async def copy_file(src: str, dest: str) -> None: