(feat) enables globbing from http/s fsspec filesystem

dlt/common/storages/fsspec_filesystem.py

@@ -4,6 +4,8 @@
 import mimetypes
 import pathlib
 import posixpath
+from datetime import datetime, timezone  # noqa: I251
+from email.utils import parsedate_to_datetime
 from io import BytesIO
 from typing import (
     Literal,
@@ -25,7 +27,6 @@
 
 from dlt import version
 from dlt.common.typing import TypedDict, NotRequired
-from dlt.common.pendulum import pendulum
 from dlt.common.configuration.specs import (
     GcpCredentials,
     AwsCredentials,
@@ -39,7 +40,7 @@
     FilesystemConfiguration,
     make_fsspec_url,
 )
-from dlt.common.time import ensure_pendulum_datetime_utc
+from dlt.common.time import ensure_datetime_utc
 from dlt.common.typing import DictStrAny
 from dlt.common.utils import without_none
 
@@ -52,35 +53,34 @@ class FileItem(TypedDict):
     relative_path: str
     mime_type: str
     encoding: NotRequired[str]
-    modification_date: pendulum.DateTime
-    size_in_bytes: int
+    modification_date: datetime
+    size_in_bytes: NotRequired[int]
     file_content: NotRequired[bytes]
 
 
-# Map of protocol to mtime resolver
+def _http_mtime(f: DictStrAny) -> Optional[datetime]:
+    """Reads `Last-Modified` which only `info()` responses carry, http listings have no mtime."""
+    last_modified = f.get("Last-Modified")
+    if last_modified is None:
+        return None
+    # an http date is not iso, only the email parser reads it
+    return ensure_datetime_utc(parsedate_to_datetime(last_modified))
+
+
+# Map of protocol to mtime resolver, returns None when the listing has no modification date
 # we only need to support a small finite set of protocols
-MTIME_DISPATCH = {
-    "s3": lambda f: ensure_pendulum_datetime_utc(f["LastModified"]),
-    "adl": lambda f: ensure_pendulum_datetime_utc(f["LastModified"]),
-    "az": lambda f: ensure_pendulum_datetime_utc(f["last_modified"]),
-    "gcs": lambda f: ensure_pendulum_datetime_utc(f["updated"]),
-    "hf": lambda f: ensure_pendulum_datetime_utc(f["last_commit"]["date"]),
-    "https": lambda f: cast(
-        pendulum.DateTime,
-        pendulum.parse(
-            f.get("Last-Modified", pendulum.now().isoformat()), exact=True, strict=False
-        ),
-    ),
-    "http": lambda f: cast(
-        pendulum.DateTime,
-        pendulum.parse(
-            f.get("Last-Modified", pendulum.now().isoformat()), exact=True, strict=False
-        ),
-    ),
-    "file": lambda f: ensure_pendulum_datetime_utc(f["mtime"]),
-    "memory": lambda f: ensure_pendulum_datetime_utc(f["created"]),
-    "gdrive": lambda f: ensure_pendulum_datetime_utc(f["modifiedTime"]),
-    "sftp": lambda f: ensure_pendulum_datetime_utc(f["mtime"]),
+MTIME_DISPATCH: Dict[str, Callable[[DictStrAny], Optional[datetime]]] = {
+    "s3": lambda f: ensure_datetime_utc(f["LastModified"]),
+    "adl": lambda f: ensure_datetime_utc(f["LastModified"]),
+    "az": lambda f: ensure_datetime_utc(f["last_modified"]),
+    "gcs": lambda f: ensure_datetime_utc(f["updated"]),
+    "hf": lambda f: ensure_datetime_utc(f["last_commit"]["date"]),
+    "https": _http_mtime,
+    "http": _http_mtime,
+    "file": lambda f: ensure_datetime_utc(f["mtime"]),
+    "memory": lambda f: ensure_datetime_utc(f["created"]),
+    "gdrive": lambda f: ensure_datetime_utc(f["modifiedTime"]),
+    "sftp": lambda f: ensure_datetime_utc(f["mtime"]),
 }
 # Support aliases
 MTIME_DISPATCH["gs"] = MTIME_DISPATCH["gcs"]
@@ -350,14 +350,19 @@ def guess_mime_type(file_name: str) -> Sequence[str]:
 
 
 def glob_files(
-    fs_client: AbstractFileSystem, bucket_url: str, file_glob: str = "**"
+    fs_client: AbstractFileSystem,
+    bucket_url: str,
+    file_glob: str = "**",
+    fetch_file_info: bool = False,
 ) -> 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.
+        fetch_file_info (bool): Calls `info` on each listed file that the listing reports no size
+            or modification date for, at the cost of one request per file. Defaults to False.
 
     Returns:
         Iterable[FileItem]: The list of files.
@@ -387,10 +392,18 @@ def glob_files(
                 " version 2023.9.0 or later"
             )
 
+    scheme = bucket_url_parsed.scheme
+
     for file, md in glob_result.items():
         if md["type"] != "file":
             continue
-        scheme = bucket_url_parsed.scheme
+        size, modification_date = md.get("size"), MTIME_DISPATCH[scheme](md)
+        # if mtime or size are not available, get info for particular file
+        if fetch_file_info and (size is None or modification_date is None):
+            md = fs_client.info(file)
+            size, modification_date = md.get("size"), MTIME_DISPATCH[scheme](md)
+        if modification_date is None:
+            modification_date = datetime.now(timezone.utc)
 
         # relative paths are always POSIX
         if is_local_fs:
@@ -410,9 +423,10 @@ def glob_files(
             relative_path=rel_path,
             file_url=file_url,
             mime_type=mime_type,
-            modification_date=MTIME_DISPATCH[scheme](md),
-            size_in_bytes=int(md["size"]),
+            modification_date=modification_date,
         )
+        if size is not None:
+            file_item["size_in_bytes"] = int(size)
         if encoding is not None:
             file_item["encoding"] = encoding
         yield file_item

dlt/common/storages/transactional_file.py

@@ -9,13 +9,13 @@
 import string
 import time
 import typing as t
+from datetime import datetime, timedelta, timezone  # noqa: I251
 from pathlib import Path
 import posixpath
 from contextlib import contextmanager
 from threading import Timer
 import fsspec
 
-from dlt.common.pendulum import pendulum, timedelta
 from dlt.common.storages.fsspec_filesystem import MTIME_DISPATCH
 
 
@@ -96,14 +96,22 @@ def _stop_heartbeat(self) -> None:
     def _sync_locks(self) -> t.List[str]:
         """Gets a list of lock names after removing stale locks. The list is time-sortable with earliest created lock coming first."""
         output = []
-        now = pendulum.now()
+        now = datetime.now(timezone.utc)
 
         for lock in self._fs.ls(posixpath.dirname(self.lock_path), refresh=True, detail=True):
             name = lock["name"]
             if not name.startswith(self.lock_prefix):
                 continue
-            # Purge stale locks
+            # purge stale locks
             mtime = self.extract_mtime(lock)
+            if mtime is None:
+                # without an mtime a lock left behind by a crashed holder never expires and
+                # everybody else waits on it forever
+                raise RuntimeError(
+                    f"Filesystem `{self._fs.protocol}` reports no modification time for lock file"
+                    f" `{name}` so stale locks cannot be detected. `TransactionalFile` cannot lock"
+                    " on this filesystem."
+                )
             if now - mtime > timedelta(seconds=TransactionalFile.LOCK_TTL_SECONDS):
                 try:  # Janitors can race, so we ignore errors
                     self._fs.rm(name)

dlt/sources/filesystem/__init__.py

@@ -103,6 +103,7 @@ def filesystem(  # noqa DOC
     kwargs: Optional[Dict[str, Any]] = None,
     client_kwargs: Optional[Dict[str, Any]] = None,
     incremental: Optional[dlt.sources.incremental[Any]] = None,
+    fetch_file_info: bool = False,
 ) -> Iterator[List[FileItem]]:
     """This resource lists files in `bucket_url` using `file_glob` pattern. The files are yielded as FileItem which also
     provide methods to open and read file data. It should be combined with transformers that further process (ie. load files)
@@ -118,6 +119,8 @@ def filesystem(  # noqa DOC
         client_kwargs (Optional[Dict[str, Any]]): Additional arguments passed to underlying fsspec native client ie. dict(verify="public.crt) for botocore
         incremental (Optional[dlt.sources.incremental[Any]]): Defines incremental cursor on listed files, with `modification_date`
             being the most common choice that returns only files created from the previous run.
+        fetch_file_info (bool, optional): If true, `dlt` gets missing `size_in_bytes` and `modification_date` from file details. This
+            requires a server call per file. Currently needed by `http` filesystem only. Defaults to False.
 
     Yields:
         List[FileItem]: The list of files.
@@ -133,7 +136,7 @@ def filesystem(  # noqa DOC
 
     files_chunk: List[FileItem] = []
 
-    iter_ = glob_files(fs_client, bucket_url, file_glob)
+    iter_ = glob_files(fs_client, bucket_url, file_glob, fetch_file_info)
 
     # if incremental is set with row order, use it to order the results
     # NOTE: fsspec glob for buckets reads all files before running iterator
@@ -144,7 +147,7 @@ def filesystem(  # noqa DOC
         )
         iter_ = iter(
             sorted(
-                list(glob_files(fs_client, bucket_url, file_glob)),
+                list(glob_files(fs_client, bucket_url, file_glob, fetch_file_info)),
                 key=lambda f_: f_[incremental.cursor_path],  # type: ignore[literal-required]
                 reverse=reverse,
             )

dlt/sources/filesystem/helpers.py

@@ -29,6 +29,7 @@ class FilesystemConfigurationResource(FilesystemConfigurationWithLocalFiles):
     file_glob: Optional[str] = "*"
     files_per_page: int = DEFAULT_CHUNK_SIZE
     extract_content: bool = False
+    fetch_file_info: bool = False
 
     @resolve_type("credentials")
     def resolve_credentials_type(self) -> Type[CredentialsConfiguration]:

docs/website/docs/dlt-ecosystem/verified-sources/filesystem/index.md

@@ -366,6 +366,15 @@ Full list of `filesystem` resource parameters:
 
 * `files_per_page` - number of files processed at once. The default value is `100`.
 * `extract_content` - if true, the content of the file will be read and returned in the resource. The default value is `False`.
+* `fetch_file_info` - if true, dlt reads `size_in_bytes` and `modification_date` that the listing does not
+  report. The default value is `False`.
+
+  :::note
+  Only the `http` and `https` filesystems need this parameter. HTTP has no listing protocol. fsspec builds
+  the file list from the HTML index page, which reports neither size nor modification date. Without
+  `fetch_file_info`, these files have no `size_in_bytes`. Their `modification_date` is the time of the
+  listing. With `fetch_file_info`, dlt makes one extra request per file to read both values.
+  :::
 
 ### 2. Choose the right reader
 
@@ -607,6 +616,11 @@ filtered_files = filesystem(bucket_url="s3://bucket_name", file_glob="**/*.json"
 
 If for some reason you only want to load small files, you can also do that:
 
+:::note
+Over `http` and `https`, the resource needs `fetch_file_info=True`. Without `fetch_file_info`,
+`size_in_bytes` is missing and the filter below raises `KeyError`.
+:::
+
 ```py
 import dlt
 from dlt.sources.filesystem import filesystem, read_csv
@@ -653,8 +667,8 @@ The filesystem ensures consistent file representation across bucket types and of
 - `file_name` - name of the file from the bucket URL.
 - `relative_path` - set when doing `glob`, is a relative path to a `bucket_url` argument.
 - `mime_type` - file's MIME type. It is sourced from the bucket provider or inferred from its extension.
-- `modification_date` - file's last modification time (format: `pendulum.DateTime`).
-- `size_in_bytes` - file size.
+- `modification_date` - file's last modification time (format: `datetime.datetime`, always UTC).
+- `size_in_bytes` - file size. Not present when the filesystem does not report it (see `fetch_file_info`).
 - `file_content` - content, provided upon request.
 
 :::info

tests/common/schema/test_schema_utils.py

@@ -174,13 +174,15 @@ def test_version_table_has_column_descriptions() -> None:
         "inserted_at",
         "schema_name",
         "version_hash",
-        "schema"
+        "schema",
     ]
     for col_name in expected_columns:
         assert col_name in table["columns"], f"Column {col_name} missing from version_table"
         col = table["columns"][col_name]
         assert "description" in col, f"Column {col_name} missing description"
-        assert isinstance(col["description"], str), f"Column {col_name} description should be a string"
+        assert isinstance(
+            col["description"], str
+        ), f"Column {col_name} description should be a string"
         assert len(col["description"]) > 0, f"Column {col_name} description should not be empty"
 
 
@@ -193,18 +195,14 @@ def test_loads_table_has_column_descriptions() -> None:
     assert table["description"] == "Created by DLT. Tracks completed loads"
 
     # Verify all columns have descriptions
-    expected_columns = [
-        "load_id",
-        "schema_name",
-        "status",
-        "inserted_at",
-        "schema_version_hash"
-    ]
+    expected_columns = ["load_id", "schema_name", "status", "inserted_at", "schema_version_hash"]
     for col_name in expected_columns:
         assert col_name in table["columns"], f"Column {col_name} missing from loads_table"
         col = table["columns"][col_name]
         assert "description" in col, f"Column {col_name} missing description"
-        assert isinstance(col["description"], str), f"Column {col_name} description should be a string"
+        assert isinstance(
+            col["description"], str
+        ), f"Column {col_name} description should be a string"
         assert len(col["description"]) > 0, f"Column {col_name} description should not be empty"
 
 
@@ -231,7 +229,9 @@ def test_pipeline_state_table_has_column_descriptions() -> None:
         assert col_name in table["columns"], f"Column {col_name} missing from pipeline_state_table"
         col = table["columns"][col_name]
         assert "description" in col, f"Column {col_name} missing description"
-        assert isinstance(col["description"], str), f"Column {col_name} description should be a string"
+        assert isinstance(
+            col["description"], str
+        ), f"Column {col_name} description should be a string"
         assert len(col["description"]) > 0, f"Column {col_name} description should not be empty"
 
 
@@ -257,9 +257,9 @@ def test_internal_tables_column_descriptions_are_not_empty() -> None:
 
     for table_name, table in tables:
         for col_name, col in table["columns"].items():
-            assert "description" in col, (
-                f"Table {table_name}, column {col_name} missing description"
-            )
-            assert col["description"] is not None and col["description"].strip() != "", (
-                f"Table {table_name}, column {col_name} has empty description"
-            )
+            assert (
+                "description" in col
+            ), f"Table {table_name}, column {col_name} missing description"
+            assert (
+                col["description"] is not None and col["description"].strip() != ""
+            ), f"Table {table_name}, column {col_name} has empty description"

tests/common/storages/test_transactional_file.py

@@ -185,3 +185,13 @@ def test_file_transaction_directory(fs: fsspec.AbstractFileSystem):
             writer.write(b"test 1")
 
         writer.release_lock()
+
+
+def test_lock_requires_mtime(fs: fsspec.AbstractFileSystem, file_name: str):
+    """A filesystem without mtime cannot expire stale locks, so locking must fail loudly
+    instead of waiting on a lock no one will ever release."""
+    writer = TransactionalFile(file_name, fs)
+    writer.extract_mtime = lambda _: None
+
+    with pytest.raises(RuntimeError, match="no modification time"):
+        writer.acquire_lock()

tests/common/storages/utils.py

@@ -1,5 +1,6 @@
 import os
 import glob
+from datetime import datetime  # noqa: I251
 from pathlib import Path
 from urllib.parse import urlparse
 import pytest
@@ -88,7 +89,7 @@ def assert_sample_files(
         assert item["file_url"].endswith(item["relative_path"])
         assert isinstance(item["mime_type"], str)
         assert isinstance(item["size_in_bytes"], int)
-        assert isinstance(item["modification_date"], pendulum.DateTime)
+        assert isinstance(item["modification_date"], datetime)
 
         # create file dict
         file_dict = FileItemDict(item, fs_client)

tests/load/filesystem/test_filesystem_common.py

@@ -134,7 +134,9 @@ def check_file_exists(filedir_: str, file_url_: str):
     def check_file_changed(file_url_: str):
         details = filesystem.info(file_url_)
         assert details["size"] == 11
-        assert (MTIME_DISPATCH[config.protocol](details) - now).seconds < 160
+        mtime = MTIME_DISPATCH[config.protocol](details)
+        assert mtime is not None
+        assert (mtime - now).seconds < 160
 
     bucket_url = os.environ["DESTINATION__FILESYSTEM__BUCKET_URL"]
     config = get_config()

tests/load/sources/filesystem/test_filesystem_source.py

@@ -1,4 +1,5 @@
 import os
+from datetime import datetime  # noqa: I251
 from typing import Any, Dict, List, cast
 
 from fsspec import AbstractFileSystem
@@ -68,7 +69,7 @@ def assert_sample_content(items: List[FileItemDict]):
             assert item["size_in_bytes"] == 14
             assert item["file_url"].endswith("/samples/sample.txt")
             assert item["mime_type"] == "text/plain"
-            assert isinstance(item["modification_date"], pendulum.DateTime)
+            assert isinstance(item["modification_date"], datetime)
 
         yield items