fix(filesystem): list files through every registered scheme

docs/changelog.md

@@ -2,6 +2,11 @@
 
 ## in progress
 
+- Filesystem: List and read through every registered scheme, including `r2://`,
+  `oss://`, `hdfs://`, `smb://`, `ftp://`, `dbfs://`, `oci://` and `webhdfs://`,
+  by resolving a file's modification date from the listing the filesystem client
+  returns.
+
 ## 2026/08/04 v0.9.0
 
 - Database: Load SQLite and DuckDB databases that live on a remote filesystem,

src/dlt_filesystem/source/adapter.py

@@ -19,7 +19,7 @@
 import dlt
 from dlt.sources import DltResource
 from dlt.sources.credentials import FileSystemCredentials
-from dlt.sources.filesystem import FileItem, FileItemDict, fsspec_filesystem, glob_files
+from dlt.sources.filesystem import FileItem, FileItemDict, fsspec_filesystem
 from fsspec import AbstractFileSystem
 
 from dlt_filesystem.source.error import NoFilesFoundError
@@ -38,6 +38,7 @@
     read_xml,
     read_yaml,
 )
+from dlt_filesystem.source.lister import glob_files
 
 from .model import FilesystemConfigurationResource
 

src/dlt_filesystem/source/lister.py

@@ -0,0 +1,178 @@
+"""File discovery for the filesystem sources.
+
+``glob_files`` is vendored from dlt (Apache-2.0,
+``dlt.common.storages.fsspec_filesystem``) so a file's modification date is
+resolved from the shape the filesystem client actually returns.
+
+dlt selects that value through a table keyed by URL scheme, where each entry
+reads one backend-specific key out of the listing (``LastModified`` for ``s3``,
+``updated`` for ``gs``, ``last_modified`` for ``az``). Both halves of that
+coupling break here:
+
+- A scheme dlt does not know raises ``KeyError`` on the scheme itself, so
+  listing crashes on the first matched file for ``r2://``, ``oss://``,
+  ``hdfs://``, ``smb://``, ``ftp://``, ``dbfs://``, ``oci://`` and
+  ``webhdfs://``.
+- A client whose listing uses a different key than its scheme is mapped to
+  raises ``KeyError`` on the key, which is what any ``pyarrow.fs``-backed
+  client does (its entries carry ``mtime`` whatever the scheme).
+
+Resolving from the listing keeps every scheme dlt does know byte-identical,
+because its own table is consulted first.
+"""
+
+import glob
+import pathlib
+import posixpath
+from datetime import datetime, timezone
+from typing import Any, Callable, Iterator, Mapping, Tuple
+from urllib.parse import urlparse
+
+from dlt.common.storages import FilesystemConfiguration
+from dlt.common.storages.configuration import make_fsspec_url
+from dlt.common.storages.fsspec_filesystem import (
+    MTIME_DISPATCH,
+    FileItem,
+    guess_mime_type,
+)
+from dlt.common.time import ensure_pendulum_datetime_utc
+from fsspec import AbstractFileSystem
+
+
+def _from_epoch_millis(value: Any) -> Any:
+    """Read a timestamp in milliseconds, as WebHDFS reports one."""
+    return ensure_pendulum_datetime_utc(float(value) / 1000)
+
+
+def _from_mlsd_timestamp(value: Any) -> Any:
+    """Read an RFC 3659 ``modify`` fact, ``YYYYMMDDHHMMSS[.sss]``, as FTP reports one.
+
+    fsspec falls back to parsing ``dir`` output on servers without MLSD, which
+    yields a year-less ``Aug 4 09:30``. That cannot name an instant, so it is
+    rejected here and surfaces as the "no usable modification date" error.
+    """
+    stamp = datetime.strptime(str(value)[:14], "%Y%m%d%H%M%S")
+    return ensure_pendulum_datetime_utc(stamp.replace(tzinfo=timezone.utc))
+
+
+# The key each backend's listing carries a file's last-modified time under,
+# paired with how that backend encodes it. Every entry is taken from the
+# backend's own `modified()` implementation, which reads the same key.
+MODIFICATION_DATE_KEYS: Tuple[Tuple[str, Callable[[Any], Any]], ...] = (
+    ("LastModified", ensure_pendulum_datetime_utc),  # s3fs, ossfs
+    ("last_modified", ensure_pendulum_datetime_utc),  # adlfs
+    ("updated", ensure_pendulum_datetime_utc),  # gcsfs
+    ("modifiedTime", ensure_pendulum_datetime_utc),  # Google Drive
+    ("modificationTime", _from_epoch_millis),  # WebHDFS
+    ("timeModified", ensure_pendulum_datetime_utc),  # ocifs
+    ("modified", ensure_pendulum_datetime_utc),  # fsspec-databricks
+    ("modify", _from_mlsd_timestamp),  # FTP
+    # `mtime` last: SMB carries both it and `time`, where `time` is the access
+    # time, so a backend that offers a more specific key is preferred first.
+    ("mtime", ensure_pendulum_datetime_utc),  # local, SMB, pyarrow.fs clients
+)
+
+
+def resolve_modification_date(scheme: str, file_info: Mapping[str, Any]) -> Any:
+    """Return a file's modification date from one entry of a filesystem listing.
+
+    Prefers dlt's own per-scheme extractor so known schemes keep their exact
+    behaviour, then reads the key the backend emits and decodes it the way that
+    backend encodes it.
+
+    Raises:
+        ValueError: when the listing carries no usable modification date, naming
+            the scheme, the keys present, and why any candidate was rejected.
+    """
+    extractor = MTIME_DISPATCH.get(scheme)
+    if extractor is not None:
+        try:
+            return extractor(file_info)
+        except KeyError:
+            # The scheme is known but this client reports a different key, e.g.
+            # any pyarrow.fs filesystem addressed as `s3://`.
+            pass
+
+    rejected = []
+    for key, decode in MODIFICATION_DATE_KEYS:
+        if key not in file_info:
+            continue
+        try:
+            return decode(file_info[key])
+        except (ValueError, TypeError, OverflowError) as ex:
+            rejected.append(f"{key}={file_info[key]!r} ({ex})")
+
+    detail = f" Rejected: {'; '.join(rejected)}." if rejected else ""
+    raise ValueError(
+        f"Filesystem listing for scheme '{scheme}' carries no usable modification "
+        f"date. Keys present: {sorted(file_info)}.{detail}"
+    )
+
+
+def glob_files(
+    fs_client: AbstractFileSystem, bucket_url: str, file_glob: str = "**"
+) -> Iterator[FileItem]:
+    """Get the files from the filesystem client.
+
+    Args:
+        fs_client (AbstractFileSystem): The filesystem client.
+        bucket_url (str): The url to the bucket.
+        file_glob (str): A glob for the filename filter.
+
+    Returns:
+        Iterable[FileItem]: The list of files.
+    """
+    is_local_fs = "file" in fs_client.protocol
+    if is_local_fs and FilesystemConfiguration.is_local_path(bucket_url):
+        bucket_url = FilesystemConfiguration.make_file_url(bucket_url)
+    bucket_url_parsed = urlparse(bucket_url)
+
+    if is_local_fs:
+        root_dir = FilesystemConfiguration.make_local_path(bucket_url)
+        # use a Python glob to get files
+        files = glob.glob(
+            str(pathlib.Path(root_dir).joinpath(file_glob)), recursive=True
+        )
+        glob_result = {file: fs_client.info(file) for file in files}
+    else:
+        # convert to fs_path
+        root_dir = fs_client._strip_protocol(bucket_url)
+        filter_url = posixpath.join(root_dir, file_glob)
+        # dlt's copy invalidates the listing cache for the `hf` protocol here.
+        # No HuggingFace scheme is registered, so that branch is left out.
+        glob_result = fs_client.glob(filter_url, detail=True)
+        if isinstance(glob_result, list):
+            raise NotImplementedError(
+                "Cannot request details when using fsspec.glob. For adlfs (Azure)"
+                " please use version 2023.9.0 or later"
+            )
+
+    for file, md in glob_result.items():
+        if md["type"] != "file":
+            continue
+        scheme = bucket_url_parsed.scheme
+
+        # relative paths are always POSIX
+        if is_local_fs:
+            # use OS pathlib for local paths
+            loc_path = pathlib.Path(file)
+            file_name = loc_path.name
+            rel_path = loc_path.relative_to(root_dir).as_posix()
+            file_url = FilesystemConfiguration.make_file_url(file)
+        else:
+            file_name = posixpath.basename(file)
+            rel_path = posixpath.relpath(file, root_dir)
+            file_url = make_fsspec_url(scheme, file, bucket_url)
+
+        mime_type, encoding = guess_mime_type(rel_path)
+        file_item = FileItem(
+            file_name=file_name,
+            relative_path=rel_path,
+            file_url=file_url,
+            mime_type=mime_type,
+            modification_date=resolve_modification_date(scheme, md),
+            size_in_bytes=int(md["size"]),
+        )
+        if encoding is not None:
+            file_item["encoding"] = encoding
+        yield file_item

tests/dlt_filesystem/test_source_lister.py

@@ -0,0 +1,127 @@
+import datetime as dt
+
+import pytest
+from dlt.common.storages.fsspec_filesystem import MTIME_DISPATCH
+from fsspec import AbstractFileSystem
+
+from dlt_filesystem.source.lister import glob_files, resolve_modification_date
+
+MODIFIED = dt.datetime(2026, 8, 4, 9, 30, tzinfo=dt.timezone.utc)
+EPOCH_SECONDS = MODIFIED.timestamp()
+
+
+class StubFilesystem(AbstractFileSystem):
+    """A filesystem client that lists one file with a caller-supplied shape."""
+
+    def __init__(self, file_info: dict):
+        super().__init__()
+        self.file_info = file_info
+
+    def _strip_protocol(self, url: str) -> str:  # ty: ignore[invalid-method-override]
+        return url.split("://", 1)[-1].rstrip("/")
+
+    def glob(self, path, maxdepth=None, **kwargs) -> dict:
+        return {"bucket/data/report.csv": self.file_info}
+
+
+def stub_filesystem(protocol: str, file_info: dict) -> StubFilesystem:
+    """Return a stub client that reports `protocol` as its fsspec protocol."""
+    return type("StubFilesystem", (StubFilesystem,), {"protocol": protocol})(file_info)
+
+
+def listing(**overrides) -> dict:
+    """One entry of a filesystem listing, minus its modification-date key."""
+    return {"name": "bucket/data/report.csv", "size": 12, "type": "file", **overrides}
+
+
+@pytest.mark.parametrize(
+    "scheme,key,raw",
+    [
+        ("s3", "LastModified", MODIFIED),
+        ("s3a", "LastModified", MODIFIED),
+        ("gs", "updated", MODIFIED),
+        ("gcs", "updated", MODIFIED),
+        ("az", "last_modified", MODIFIED),
+        ("abfss", "last_modified", MODIFIED),
+        ("file", "mtime", EPOCH_SECONDS),
+        ("gdrive", "modifiedTime", MODIFIED),
+    ],
+)
+def test_known_scheme_keeps_dlt_extractor(scheme, key, raw):
+    """Schemes dlt knows resolve through dlt's own per-scheme extractor."""
+    assert scheme in MTIME_DISPATCH
+    assert resolve_modification_date(scheme, listing(**{key: raw})) == MODIFIED
+
+
+@pytest.mark.parametrize(
+    "scheme,key,raw",
+    [
+        # Each value is the shape the backend's own `modified()` reads.
+        ("r2", "LastModified", MODIFIED),  # s3fs
+        ("oss", "LastModified", MODIFIED),  # ossfs
+        ("hdfs", "mtime", MODIFIED),  # pyarrow.fs via ArrowFSWrapper
+        ("smb", "mtime", EPOCH_SECONDS),  # os.stat_result.st_mtime
+        ("ftp", "modify", "20260804093000"),  # RFC 3659 MLSD fact
+        ("dbfs", "modified", MODIFIED),  # fsspec-databricks
+        ("oci", "timeModified", MODIFIED),  # ocifs
+        ("webhdfs", "modificationTime", int(EPOCH_SECONDS * 1000)),  # epoch millis
+    ],
+)
+def test_scheme_dlt_does_not_know_resolves_from_the_listing(scheme, key, raw):
+    """Schemes absent from dlt's table resolve from the key their backend emits."""
+    assert scheme not in MTIME_DISPATCH
+    assert resolve_modification_date(scheme, listing(**{key: raw})) == MODIFIED
+
+
+def test_known_scheme_with_foreign_key_falls_back():
+    """A pyarrow.fs client addressed as `s3://` reports `mtime`, not `LastModified`."""
+    assert resolve_modification_date("s3", listing(mtime=MODIFIED)) == MODIFIED
+
+
+def test_access_time_is_not_read_as_a_modification_date():
+    """SMB carries both `mtime` and `time`, where `time` is the access time."""
+    accessed = dt.datetime(2026, 8, 4, 18, 0, tzinfo=dt.timezone.utc)
+    info = listing(mtime=EPOCH_SECONDS, time=accessed.timestamp())
+
+    assert resolve_modification_date("smb", info) == MODIFIED
+
+
+def test_listing_without_a_modification_date_names_what_it_saw():
+    with pytest.raises(ValueError) as excinfo:
+        resolve_modification_date("acme", listing())
+    message = str(excinfo.value)
+    assert "'acme'" in message
+    assert "'name', 'size', 'type'" in message
+
+
+def test_year_less_ftp_timestamp_is_rejected_with_its_value():
+    """fsspec parses `dir` output into a year-less `modify` on servers without MLSD."""
+    with pytest.raises(ValueError) as excinfo:
+        resolve_modification_date("ftp", listing(modify="Aug 4 09:30"))
+    message = str(excinfo.value)
+    assert "no usable modification date" in message
+    assert "modify='Aug 4 09:30'" in message
+
+
+@pytest.mark.parametrize("scheme", ["s3", "r2", "oss"])
+def test_glob_files_lists_s3_compatible_schemes(scheme):
+    """Listing an S3-compatible bucket yields files whatever scheme addresses it."""
+    fs = stub_filesystem(scheme, listing(LastModified=MODIFIED))
+
+    files = list(glob_files(fs, f"{scheme}://bucket/data", "*.csv"))
+
+    assert len(files) == 1
+    assert files[0]["file_name"] == "report.csv"
+    assert files[0]["relative_path"] == "report.csv"
+    assert files[0]["file_url"] == f"{scheme}://bucket/data/report.csv"
+    assert files[0]["modification_date"] == MODIFIED
+    assert files[0]["size_in_bytes"] == 12
+
+
+def test_glob_files_lists_arrow_backed_clients():
+    """A pyarrow.fs-backed client lists through the same path as its fsspec peer."""
+    fs = stub_filesystem("s3", listing(mtime=MODIFIED))
+
+    files = list(glob_files(fs, "s3://bucket/data", "*.csv"))
+
+    assert [file["modification_date"] for file in files] == [MODIFIED]

tests/main/filesystem/test_remote_integration.py

@@ -295,6 +295,28 @@ def test_remote_glob_concatenates_matching_objects(remote_filesystem, tmp_path):
     assert duckdb_table_cardinality(db_path, "testdrive.data") == 4
 
 
+def test_r2_scheme_reads_s3_compatible_storage(s3_emulator, tmp_path):
+    """`r2://` lists and reads, though dlt's own listing knows no such scheme.
+
+    R2 speaks the S3 API, so the S3 emulator serves it; the scheme is what this
+    covers, because listing resolves each file's modification date per scheme.
+    """
+    s3_emulator.upload("r2/part-1.csv", CSV_ROWS)
+    s3_emulator.upload("r2/part-2.csv", CSV_ROWS)
+
+    db_path = tmp_path / "r2.duckdb"
+    result = run_ingest(
+        source_uri=s3_emulator.source_uri.replace("s3://", "r2://", 1),
+        dest_uri=f"duckdb:///{db_path}",
+        source_table=s3_emulator.source_table("r2/*.csv"),
+        dest_table="testdrive.data",
+        progress="log",
+    )
+
+    assert result is not None
+    assert duckdb_table_cardinality(db_path, "testdrive.data") == 4
+
+
 @pytest.mark.parametrize("remote_filesystem", REMOTE_BACKENDS, indirect=True)
 def test_remote_unmatched_glob_succeeds(remote_filesystem, tmp_path):
     """A wildcard with no remote matches remains a valid empty selection."""