add remote HTTP regression tests

dascore/config.py

@@ -99,6 +99,10 @@ class DascoreConfig(BaseModel):
         default=1_048_576,
         description="Block size in bytes for general remote file downloads.",
     )
+    remote_download_timeout: float = Field(
+        default=60.0,
+        description="Timeout in seconds for blocking remote file downloads.",
+    )
     remote_hdf5_block_size: int = Field(
         default=5_242_880,
         description="Block size in bytes for remote HDF5 access on tuned protocols.",

dascore/utils/hdf5.py

@@ -48,7 +48,8 @@
     list_ser_to_str,
 )
 from dascore.utils.remote_io import (
-    FallbackFileObj,
+    _FallbackFileObj,
+    _get_cached_local_file,
     ensure_local_file,
     get_local_handle,
     is_no_range_http_error,
@@ -63,6 +64,130 @@
 ns_to_timedelta = partial(pd.to_timedelta, unit="ns")
 
 
+class _ManagedH5pyFile:
+    """
+    DASCore's internal h5py handle wrapper with deterministic close behavior.
+
+    All h5py-backed DASCore reads return this wrapper so callers see one handle
+    type regardless of whether the underlying resource came from:
+    - a local path
+    - an existing h5py handle
+    - a Python file object
+    - a remote ``UPath`` opened through the fallback fileobj path
+
+    For path-backed opens, this wrapper owns only the h5py handle. For
+    ``h5py.File(..., driver="fileobj")`` paths, it also owns the Python
+    file-like object DASCore created on behalf of the caller. ``close()`` is
+    therefore the point where DASCore tears down the entire HDF5 access stack.
+    """
+
+    def __init__(self, handle: H5pyFile, owned_fileobj=None):
+        self._handle = handle
+        self._owned_fileobj = owned_fileobj
+        self._closed = False
+
+    def close(self):
+        """Close the h5py file and, when present, the owned file object."""
+        if self._closed:
+            return
+        try:
+            self._handle.close()
+        finally:
+            if self._owned_fileobj is not None:
+                with suppress(Exception):
+                    self._owned_fileobj.close()
+            self._closed = True
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        self.close()
+        return False
+
+    def __getitem__(self, item):
+        return self._handle[item]
+
+    def __contains__(self, item):
+        return item in self._handle
+
+    def __iter__(self):
+        return iter(self._handle)
+
+    @property
+    def closed(self):
+        """Return True when close has been called on the proxy."""
+        return self._closed
+
+    def __getattr__(self, item):
+        return getattr(self._handle, item)
+
+
+def open_h5_resource(
+    resource,
+    *,
+    mode: str,
+    constructor,
+    open_kwargs_getter,
+) -> _ManagedH5pyFile:
+    """
+    Open an HDF5 resource and return DASCore's managed h5py handle wrapper.
+
+    This is the central constructor for h5py-backed reads in DASCore. It keeps
+    the branching needed for local paths, already-open handles, remote
+    fileobj-backed reads, cached-local reuse, and no-range HTTP fallback in one
+    place so ``H5Reader.get_handle()`` stays thin.
+
+    Parameters
+    ----------
+    resource
+        A local path, remote ``UPath``, open file object, or existing h5py
+        handle.
+    mode
+        The mode to pass to the h5py constructor.
+    constructor
+        The callable used to construct an h5py handle.
+    open_kwargs_getter
+        Callback which returns backend-specific kwargs for remote file opens.
+    """
+    if isinstance(resource, _ManagedH5pyFile):
+        return resource
+    if isinstance(resource, H5pyFile):
+        return _ManagedH5pyFile(resource)
+    if isinstance(resource, io.IOBase):
+        handle = constructor(resource, mode=mode, driver="fileobj")
+        return _ManagedH5pyFile(handle, resource)
+    if isinstance(resource, UPath):
+        # Reuse an already-materialized local artifact when present so later
+        # HDF5 reads do not re-enter the remote fallback path unnecessarily.
+        if cached_path := _get_cached_local_file(resource):
+            return open_h5_resource(
+                cached_path,
+                mode=mode,
+                constructor=constructor,
+                open_kwargs_getter=open_kwargs_getter,
+            )
+        file_mode = "rb" if mode == "r" else "r+b"
+        open_kwargs = open_kwargs_getter(resource)
+        handle = _FallbackFileObj(
+            remote_opener=lambda: resource.open(file_mode, **open_kwargs),
+            local_opener=lambda: ensure_local_file(resource).open(file_mode),
+            error_predicate=is_no_range_http_error,
+        )
+        try:
+            h5_handle = constructor(handle, mode=mode, driver="fileobj")
+            return _ManagedH5pyFile(h5_handle, handle)
+        except Exception:
+            handle.close()
+            raise
+    try:
+        _maybe_make_parent_directory(resource)
+        return _ManagedH5pyFile(constructor(resource, mode=mode))
+    except TypeError:
+        msg = f"Couldn't get handle from {resource} using h5py"
+        raise NotImplementedError(msg)
+
+
 class _HDF5Store(pd.HDFStore):
     """
     A work-around for pandas HDF5 store not accepting
@@ -182,7 +307,7 @@ class HDFPatchIndexManager:
         }
     )
     # functions to apply to decode dataframe after loading from hdf file
-    _column_decorders = FrozenDict(
+    _column_decoders = FrozenDict(
         {
             "time_min": ns_to_datetime,
             "time_max": ns_to_datetime,
@@ -250,7 +375,7 @@ def encode_table(self, df, path=None):
     def decode_table(self, df):
         """Decode the table from hdf5."""
         # ensure the base path is not in the path column
-        for col, func in self._column_decorders.items():
+        for col, func in self._column_decoders.items():
             df[col] = func(df[col])
         # populate index store and update metadata
         # assert not df.isnull().any().any(), "null values found in index"
@@ -495,24 +620,14 @@ def get_handle(cls, resource):
         Unlike PyTablesReader, h5py can consume a binary file object via the
         ``fileobj`` driver, so remote UPath inputs stay streaming-based here.
         """
-        if isinstance(resource, cls | H5pyFile):
+        if isinstance(resource, cls | _ManagedH5pyFile):
             return resource
-        if isinstance(resource, io.IOBase):
-            return cls.constructor(resource, mode=cls.mode, driver="fileobj")
-        if isinstance(resource, UPath):
-            mode = "rb" if cls.mode == "r" else "r+b"
-            open_kwargs = cls._get_open_kwargs(resource)
-            handle = FallbackFileObj(
-                remote_opener=lambda: resource.open(mode, **open_kwargs),
-                local_opener=lambda: ensure_local_file(resource).open(mode),
-                error_predicate=is_no_range_http_error,
-            )
-            try:
-                return cls.constructor(handle, mode=cls.mode, driver="fileobj")
-            except Exception:
-                handle.close()
-                raise
-        return super().get_handle(resource)
+        return open_h5_resource(
+            resource,
+            mode=cls.mode,
+            constructor=cls.constructor,
+            open_kwargs_getter=cls._get_open_kwargs,
+        )
 
 
 class LocalH5Reader(H5Reader):

dascore/utils/remote_io.py

@@ -12,6 +12,7 @@
 from functools import lru_cache
 from hashlib import sha256
 from pathlib import Path
+from urllib.request import Request, urlopen
 
 from dascore.compat import UPath
 from dascore.config import get_config
@@ -125,7 +126,6 @@ def _download_remote_file(path, local_path: Path):
     """Download a remote path into its cache location."""
     resource = coerce_to_upath(path)
     protocol = getattr(resource, "protocol", None)
-    open_kwargs = {"block_size": 0} if protocol in _HTTP_PROTOCOLS else {}
     local_path.parent.mkdir(parents=True, exist_ok=True)
     fd, temp_name = tempfile.mkstemp(
         dir=local_path.parent,
@@ -135,12 +135,27 @@ def _download_remote_file(path, local_path: Path):
     os.close(fd)
     tmp_path = Path(temp_name)
     try:
-        with (
-            resource.open("rb", **open_kwargs) as remote_fi,
-            tmp_path.open("wb") as local_fi,
-        ):
-            while chunk := remote_fi.read(get_config().remote_download_block_size):
-                local_fi.write(chunk)
+        if protocol in _HTTP_PROTOCOLS:
+            # Use a direct blocking HTTP download here rather than
+            # ``resource.open(...)``. The fallback path can be entered while an
+            # active fsspec HTTP read is already in progress, and re-entering
+            # that stack from inside the fallback can deadlock.
+            headers = dict(getattr(resource, "storage_options", {}) or {})
+            # Supported public inputs are normalized to a real UPath first, so
+            # `protocol` and `str(resource)` come from the same object and do
+            # not need separate scheme validation here.
+            request = Request(str(resource), headers=headers)
+            timeout = get_config().remote_download_timeout
+            with (
+                urlopen(request, timeout=timeout) as remote_fi,
+                tmp_path.open("wb") as local_fi,
+            ):
+                while chunk := remote_fi.read(get_config().remote_download_block_size):
+                    local_fi.write(chunk)
+        else:
+            with resource.open("rb") as remote_fi, tmp_path.open("wb") as local_fi:
+                while chunk := remote_fi.read(get_config().remote_download_block_size):
+                    local_fi.write(chunk)
         tmp_path.replace(local_path)
     finally:
         tmp_path.unlink(missing_ok=True)
@@ -201,6 +216,19 @@ def ensure_local_file(resource) -> Path:
     raise TypeError(msg)
 
 
+def _get_cached_local_file(resource) -> Path | None:
+    """Return a cached local path without materializing a missing resource."""
+    if not is_pathlike(resource) or is_local_path(resource):
+        return None
+    remote = coerce_to_upath(resource)
+    cache_root = _normalize_cache_root(get_remote_cache_path())
+    remote_id = normalize_remote_id(remote)
+    local_path = (
+        cache_root / sha256(remote_id.encode()).hexdigest() / _safe_remote_name(remote)
+    )
+    return local_path if local_path.exists() else None
+
+
 def get_local_handle(resource, opener):
     """Materialize a resource locally, then pass it to an opener."""
     return opener(ensure_local_file(resource))
@@ -214,8 +242,40 @@ def is_no_range_http_error(exc: Exception) -> bool:
     )
 
 
-class FallbackFileObj:
-    """A file-like object that switches from remote to local on one error."""
+class _FallbackFileObj:
+    """
+    A seekable binary file adapter that starts remote and falls back to local.
+
+    This private wrapper is used when DASCore wants to give a consumer such as h5py a
+    normal file-like object for a remote resource without eagerly downloading
+    the whole file first.
+
+    Behavior
+    --------
+    - Opens the resource with ``remote_opener`` initially.
+    - Proxies standard file operations like ``read``, ``readinto``, ``seek``,
+      ``tell``, and ``close`` to the active handle.
+    - If one proxied operation raises an exception matched by
+      ``error_predicate``, the remote handle is abandoned and replaced with a
+      local handle from ``local_opener``.
+    - The current logical file position is preserved across that switch.
+    - Once fallback happens, all later operations use the local handle.
+
+    Why this exists
+    ---------------
+    Some remote backends work for simple sequential reads but fail when a
+    library such as h5py performs the random-access pattern required to read
+    HDF5 metadata. A common case is HTTP servers that do not support range
+    requests well enough for seek-heavy reads. This wrapper lets DASCore stay
+    remote-first when that works, while still recovering by materializing a
+    local file only when needed.
+
+    Notes
+    -----
+    This is not a general retry wrapper for arbitrary IO failures. It is meant
+    for one known fallback condition where switching from remote access to a
+    local cached file is safe and expected.
+    """
 
     def __init__(self, remote_opener, local_opener, error_predicate):
         self._remote_opener = remote_opener
@@ -239,6 +299,8 @@ def _switch_to_local(self):
         if self._using_local:
             return
         old_handle = self._handle
+        # Reopen against the stable local artifact and continue from the same
+        # logical file position the caller was already using.
         self._handle = self._local_opener()
         self._handle.seek(self._pos)
         self._using_local = True
@@ -251,6 +313,8 @@ def _with_fallback(self, func, fallback_pos=None):
         except Exception as exc:
             if not self._error_predicate(exc):
                 raise
+            # The first matching remote-read failure permanently moves this
+            # wrapper onto the local file; later operations stay local.
             self._switch_to_local()
             result = func()
             self._set_pos_from_handle(fallback=fallback_pos)
@@ -272,7 +336,7 @@ def readinto(self, b):
         return out
 
     def seek(self, offset, whence=0):
-        """Move the file cursor."""
+        """Move the file cursor, triggering fallback if random access fails."""
         out = self._with_fallback(lambda: self._handle.seek(offset, whence))
         self._set_pos_from_handle(fallback=out)
         return out

tests/test_io/test_remote_http.py

@@ -159,9 +159,12 @@ def test_http_range_hdf5_read_succeeds(
         """Range-capable HTTP servers should support DASCore HDF5 reads."""
         ensure_http_fetch_file("prodml_2.1.h5")
         path = http_range_das_path / "prodml_2.1.h5"
-        assert dc.get_format(path) == ("PRODML", "2.1")
-        assert dc.read(path)
-        assert not list(get_remote_cache_path().rglob("prodml_2.1.h5"))
+        fmt = dc.get_format(path)
+        assert fmt == ("PRODML", "2.1")
+        spool = dc.read(path)
+        assert spool
+        cached = list(get_remote_cache_path().rglob("prodml_2.1.h5"))
+        assert not cached
 
     def test_spool_file_path(self, http_regression_das_path):
         """A remote HTTP file should still produce a file-backed spool."""