Merge pull request #91 from lsst/tickets/DM-44547

DM-44547: Add fsspec constructor

Author: timj

.github/workflows/build.yaml

@@ -13,7 +13,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.11", "3.12"]
+        python-version: ["3.11", "3.12", "3.13"]
 
     steps:
       - uses: actions/checkout@v4
@@ -27,6 +27,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
           cache: "pip"
           cache-dependency-path: "setup.cfg"
+          allow-prereleases: true
 
       - name: Update pip/wheel infrastructure
         shell: bash -l {0}
@@ -46,6 +47,10 @@ jobs:
         run: |
           uv pip install --system google-cloud-storage
 
+      - name: Install fsspec for testing
+        run: |
+          uv pip install --system fsspec s3fs
+
       - name: Install dependencies
         run: |
           uv pip install --system -r requirements.txt

doc/changes/DM-44547.feature.rst

@@ -0,0 +1 @@
+Added a new method ``ResourcePath.to_fsspec()`` to return ``fsspec`` file system objects suitable for use in packages such as Astropy and Pyarrow.

doc/conf.py

@@ -12,6 +12,8 @@
 doxylink = {}
 exclude_patterns = ["changes/*"]
 
+intersphinx_mapping["fsspec"] = ("https://filesystem-spec.readthedocs.io/en/latest/", None)  # noqa: F405
+
 nitpick_ignore_regex = [
     ("py:(class|obj)", ".*ResourceHandle.U$"),
     ("py:(class|obj)", "re.Pattern"),

mypy.ini

@@ -17,6 +17,12 @@ ignore_missing_imports = True
 [mypy-wsgidav.*]
 ignore_missing_imports = True
 
+[mypy-fsspec.*]
+ignore_missing_imports = True
+
+[mypy-s3fs.*]
+ignore_missing_imports = True
+
 [mypy-defusedxml.*]
 ignore_missing_imports = True
 

python/lsst/resources/_resourcePath.py

@@ -11,6 +11,8 @@
 
 from __future__ import annotations
 
+__all__ = ("ResourcePath", "ResourcePathExpression")
+
 import concurrent.futures
 import contextlib
 import copy
@@ -26,7 +28,12 @@
 from pathlib import Path, PurePath, PurePosixPath
 from random import Random
 
-__all__ = ("ResourcePath", "ResourcePathExpression")
+try:
+    import fsspec
+    from fsspec.spec import AbstractFileSystem
+except ImportError:
+    fsspec = None
+    AbstractFileSystem = type
 
 from collections.abc import Iterable, Iterator
 from typing import TYPE_CHECKING, Any, Literal, overload
@@ -409,6 +416,21 @@ def geturl(self) -> str:
         """
         return self._uri.geturl()
 
+    def to_fsspec(self) -> tuple[AbstractFileSystem, str]:
+        """Return an abstract file system and path that can be used by fsspec.
+
+        Returns
+        -------
+        fs : `fsspec.spec.AbstractFileSystem`
+            A file system object suitable for use with the returned path.
+        path : `str`
+            A path that can be opened by the file system object.
+        """
+        if fsspec is None:
+            raise ImportError("fsspec is not available")
+        # By default give the URL to fsspec and hope.
+        return fsspec.url_to_fs(self.geturl())
+
     def root_uri(self) -> ResourcePath:
         """Return the base root URI.
 

python/lsst/resources/file.py

@@ -29,6 +29,13 @@
 from ._resourcePath import ResourcePath
 from .utils import NoTransaction, ensure_directory_is_writeable, os2posix, posix2os
 
+try:
+    import fsspec
+    from fsspec.spec import AbstractFileSystem
+except ImportError:
+    fsspec = None
+    AbstractFileSystem = type
+
 if TYPE_CHECKING:
     from .utils import TransactionProtocol
 
@@ -482,6 +489,22 @@ def _openImpl(
         with FileResourceHandle(mode=mode, log=log, uri=self, encoding=encoding) as buffer:
             yield buffer  # type: ignore
 
+    def to_fsspec(self) -> tuple[AbstractFileSystem, str]:
+        """Return an abstract file system and path that can be used by fsspec.
+
+        Returns
+        -------
+        fs : `fsspec.spec.AbstractFileSystem`
+            A file system object suitable for use with the returned path.
+        path : `str`
+            A path that can be opened by the file system object.
+        """
+        if fsspec is None:
+            raise ImportError("fsspec is not available")
+        # fsspec does not like URL encodings in file URIs so pass it the os
+        # path instead.
+        return fsspec.url_to_fs(self.ospath)
+
 
 def _create_directories(name: str | bytes) -> None:
     """Create a directory and all of its parent directories that don't yet

python/lsst/resources/packageresource.py

@@ -18,6 +18,15 @@
 import re
 from collections.abc import Iterator
 from importlib import resources
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    try:
+        import fsspec
+        from fsspec.spec import AbstractFileSystem
+    except ImportError:
+        fsspec = None
+        AbstractFileSystem = type
 
 from ._resourceHandles._baseResourceHandle import ResourceHandleProtocol
 from ._resourcePath import ResourcePath
@@ -163,3 +172,21 @@ def walk(
         for dir in dirs:
             new_uri = self.join(dir, forceDirectory=True)
             yield from new_uri.walk(file_filter)
+
+    def to_fsspec(self) -> tuple[AbstractFileSystem, str]:
+        """Return an abstract file system and path that can be used by fsspec.
+
+        Python package resources are effectively local files in most cases
+        but can be found inside ZIP files. To support this we would have
+        to change this API to a context manager (using
+        ``importlib.resources.as_file``) or find an API where fsspec knows
+        about python package resource.
+
+        Returns
+        -------
+        fs : `fsspec.spec.AbstractFileSystem`
+            A file system object suitable for use with the returned path.
+        path : `str`
+            A path that can be opened by the file system object.
+        """
+        raise NotImplementedError("fsspec can not be used with python package resources.")

python/lsst/resources/s3.py

@@ -31,6 +31,9 @@
 from ._resourceHandles._s3ResourceHandle import S3ResourceHandle
 from ._resourcePath import ResourcePath
 from .s3utils import (
+    _get_s3_connection_parameters,
+    _s3_disable_bucket_validation,
+    _s3_should_validate_bucket,
     all_retryable_errors,
     backoff,
     bucketExists,
@@ -46,6 +49,13 @@
 except ImportError:
     TransferConfig = None
 
+try:
+    import s3fs
+    from fsspec.spec import AbstractFileSystem
+except ImportError:
+    s3fs = None
+    AbstractFileSystem = type
+
 if TYPE_CHECKING:
     with contextlib.suppress(ImportError):
         import boto3
@@ -307,6 +317,33 @@ def _download_file(
             translate_client_error(err, self)
             raise
 
+    def to_fsspec(self) -> tuple[AbstractFileSystem, str]:
+        """Return an abstract file system and path that can be used by fsspec.
+
+        Returns
+        -------
+        fs : `fsspec.spec.AbstractFileSystem`
+            A file system object suitable for use with the returned path.
+        path : `str`
+            A path that can be opened by the file system object.
+        """
+        if s3fs is None:
+            raise ImportError("s3fs is not available")
+        # Must remove the profile from the URL and form it again.
+        endpoint_config = _get_s3_connection_parameters(self._profile)
+        s3 = s3fs.S3FileSystem(
+            profile=endpoint_config.profile,
+            endpoint_url=endpoint_config.endpoint_url,
+            key=endpoint_config.access_key_id,
+            secret=endpoint_config.secret_access_key,
+        )
+        if not _s3_should_validate_bucket():
+            # Accessing the s3 property forces the boto client to be
+            # constructed and cached and allows the validation to be removed.
+            _s3_disable_bucket_validation(s3.s3)
+
+        return s3, f"{self._bucket}/{self.relativeToPathRoot}"
+
     def _as_local(self) -> tuple[str, bool]:
         """Download object from S3 and place in temporary directory.
 

python/lsst/resources/s3utils.py

@@ -221,6 +221,36 @@ def getS3Client(profile: str | None = None) -> boto3.client:
     if botocore is None:
         raise ModuleNotFoundError("Could not find botocore. Are you sure it is installed?")
 
+    endpoint_config = _get_s3_connection_parameters(profile)
+
+    return _get_s3_client(endpoint_config, not _s3_should_validate_bucket())
+
+
+def _s3_should_validate_bucket() -> bool:
+    """Indicate whether bucket validation should be enabled.
+
+    Returns
+    -------
+    validate : `bool`
+        If `True` bucket names should be validated.
+    """
+    disable_value = os.environ.get("LSST_DISABLE_BUCKET_VALIDATION", "0")
+    return bool(re.search(r"^(0|f|n|false)?$", disable_value, re.I))
+
+
+def _get_s3_connection_parameters(profile: str | None = None) -> _EndpointConfig:
+    """Calculate the connection details.
+
+    Parameters
+    ----------
+    profile : `str`, optional
+        The name of an S3 profile describing which S3 service to use.
+
+    Returns
+    -------
+    config : _EndPointConfig
+        All the information necessary to connect to the bucket.
+    """
     endpoint = None
     if profile is not None:
         var_name = f"LSST_RESOURCES_S3_PROFILE_{profile}"
@@ -230,26 +260,29 @@ def getS3Client(profile: str | None = None) -> boto3.client:
     if not endpoint:
         endpoint = None  # Handle ""
 
-    disable_value = os.environ.get("LSST_DISABLE_BUCKET_VALIDATION", "0")
-    skip_validation = not re.search(r"^(0|f|n|false)?$", disable_value, re.I)
+    return _parse_endpoint_config(endpoint, profile)
+
+
+def _s3_disable_bucket_validation(client: boto3.client) -> None:
+    """Disable the bucket name validation in the client.
 
-    return _get_s3_client(endpoint, profile, skip_validation)
+    This removes the ``validate_bucket_name`` handler from the handlers
+    registered for this client.
+
+    Parameters
+    ----------
+    client : `boto3.client`
+        The client to modify.
+    """
+    client.meta.events.unregister("before-parameter-build.s3", validate_bucket_name)
 
 
 @functools.lru_cache
-def _get_s3_client(endpoint: str | None, profile: str | None, skip_validation: bool) -> boto3.client:
+def _get_s3_client(endpoint_config: _EndpointConfig, skip_validation: bool) -> boto3.client:
     # Helper function to cache the client for this endpoint
     config = botocore.config.Config(read_timeout=180, retries={"mode": "adaptive", "max_attempts": 10})
 
-    endpoint_config = _parse_endpoint_config(endpoint)
-
-    if endpoint_config.access_key_id is not None and endpoint_config.secret_access_key is not None:
-        # We already have the necessary configuration for the profile, so do
-        # not pass the profile to boto3.  boto3 will raise an exception if the
-        # profile is not defined in its configuration file, whether or not it
-        # needs to read the configuration from it.
-        profile = None
-    session = boto3.Session(profile_name=profile)
+    session = boto3.Session(profile_name=endpoint_config.profile)
 
     client = session.client(
         "s3",
@@ -259,19 +292,20 @@ def _get_s3_client(endpoint: str | None, profile: str | None, skip_validation: b
         config=config,
     )
     if skip_validation:
-        client.meta.events.unregister("before-parameter-build.s3", validate_bucket_name)
+        _s3_disable_bucket_validation(client)
     return client
 
 
 class _EndpointConfig(NamedTuple):
     endpoint_url: str | None = None
     access_key_id: str | None = None
     secret_access_key: str | None = None
+    profile: str | None = None
 
 
-def _parse_endpoint_config(endpoint: str | None) -> _EndpointConfig:
+def _parse_endpoint_config(endpoint: str | None, profile: str | None = None) -> _EndpointConfig:
     if not endpoint:
-        return _EndpointConfig()
+        return _EndpointConfig(profile=profile)
 
     parsed = parse_url(endpoint)
 
@@ -288,8 +322,18 @@ def _parse_endpoint_config(endpoint: str | None) -> _EndpointConfig:
         access_key_id = urllib.parse.unquote(access_key_id)
         secret_access_key = urllib.parse.unquote(secret_access_key)
 
+    if access_key_id is not None and secret_access_key is not None:
+        # We already have the necessary configuration for the profile, so do
+        # not pass the profile to boto3.  boto3 will raise an exception if the
+        # profile is not defined in its configuration file, whether or not it
+        # needs to read the configuration from it.
+        profile = None
+
     return _EndpointConfig(
-        endpoint_url=endpoint_url, access_key_id=access_key_id, secret_access_key=secret_access_key
+        endpoint_url=endpoint_url,
+        access_key_id=access_key_id,
+        secret_access_key=secret_access_key,
+        profile=profile,
     )
 
 

python/lsst/resources/tests.py

@@ -23,6 +23,11 @@
 from collections.abc import Iterable
 from typing import TYPE_CHECKING, Any
 
+try:
+    import fsspec
+except ImportError:
+    fsspec = None
+
 from lsst.resources import ResourcePath
 from lsst.resources.utils import makeTestTempDir, removeTestTempDir
 
@@ -882,6 +887,24 @@ def test_temporary(self) -> None:
             with ResourcePath.temporary_uri(prefix=self.root_uri, suffix="xxx/") as tmp:
                 pass
 
+    @unittest.skipIf(fsspec is None, "fsspec is not available.")
+    def test_fsspec(self) -> None:
+        """Simple read of a file."""
+        uri = self.tmpdir.join("test.txt")
+        self.assertFalse(uri.exists(), f"{uri} should not exist")
+        self.assertTrue(uri.path.endswith("test.txt"))
+
+        content = "abcdefghijklmnopqrstuv\n"
+        uri.write(content.encode())
+
+        try:
+            fs, path = uri.to_fsspec()
+        except NotImplementedError as e:
+            raise unittest.SkipTest(str(e)) from e
+        with fs.open(path, "r") as fd:
+            as_read = fd.read()
+        self.assertEqual(as_read, content)
+
     def test_open(self) -> None:
         tmpdir = ResourcePath(self.tmpdir, forceDirectory=True)
         with ResourcePath.temporary_uri(prefix=tmpdir, suffix=".txt") as tmp: