_remote

`sleap_io.io._remote` ¶

Internals for remote URL loading via fsspec.

This module contains all URL-handling primitives used by the high-level loaders in :mod:sleap_io.io.main (and the helper refactors in :mod:sleap_io.io.slp / :mod:sleap_io.io.utils). The public surface exposed to users is :func:clear_remote_cache and :class:RemoteIOError (re-exported at the package top level); everything else is private (underscore-prefixed).

Heavy third-party dependencies (fsspec, aiohttp) are imported lazily inside the functions that need them so that import sleap_io stays fast and the cheap, pure-stdlib helpers (:func:_is_url, :func:_redact_url, :func:_identify_magic) remain usable with zero heavy imports on the local hot path.

Classes:

Name	Description
`RemoteIOError`	Raised for HTTP-level failures during remote loading.

Functions:

Name	Description
`clear_remote_cache`	Clear sleap-io's fsspec remote cache.
`download`	Download a remote file to local disk.
`open_remote_h5`	Open a remote HDF5 file for membership/existence probing.
`open_url`	Open `url` as a file-like object using the configured strategy.

Attributes:

Name	Type	Description
`__annotations__`		dict() -> new empty dictionary
`__cached__`		str(object='') -> str
`__doc__`		str(object='') -> str
`__file__`		str(object='') -> str
`__name__`		str(object='') -> str
`__package__`		str(object='') -> str

`annotations = {'_FORMAT_MAGIC': 'tuple[tuple[bytes, str], ...]'}` `module-attribute` ¶

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

`cached = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/pycache/_remote.cpython-313.pyc'` `module-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

doc = 'Internals for remote URL loading via fsspec.\n\nThis module contains all URL-handling primitives used by the high-level loaders\nin :mod:`sleap_io.io.main` (and the helper refactors in\n:mod:`sleap_io.io.slp` / :mod:`sleap_io.io.utils`). The public surface exposed\nto users is :func:`clear_remote_cache` and :class:`RemoteIOError` (re-exported\nat the package top level); everything else is private (underscore-prefixed).\n\nHeavy third-party dependencies (``fsspec``, ``aiohttp``) are imported lazily\ninside the functions that need them so that ``import sleap_io`` stays fast and\nthe cheap, pure-stdlib helpers (:func:`_is_url`, :func:`_redact_url`,\n:func:`_identify_magic`) remain usable with zero heavy imports on the local\nhot path.\n' `module-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

`file = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/_remote.py'` `module-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

`name = 'sleap_io.io._remote'` `module-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

`package = 'sleap_io.io'` `module-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

`RemoteIOError` ¶

Bases: builtins.OSError

Raised for HTTP-level failures during remote loading.

Subclasses :class:OSError so that callers which already handle OSError (e.g. HDF5Video.__attrs_post_init__) degrade gracefully.

The original aiohttp/fsspec exception is intentionally not chained as __cause__ / __context__ because its string form embeds the raw URL (userinfo password, ?token= query parameter, etc.), which :func:traceback.format_exception would otherwise leak. A redacted summary of the cause (type name + redacted message) is preserved instead via :attr:cause_summary.

Attributes:

Name	Type	Description
`url`		Redacted URL (credentials stripped if present), or None.
`status`		HTTP status code, or None for connection-level errors.
`cause_summary`		Redacted, one-line summary of the underlying exception (type name plus redacted message), or None.

Methods:

Name	Description
`__init__`	Build a RemoteIOError with a redacted, composed message.

Source code in sleap_io/io/_remote.py

class RemoteIOError(OSError):
    """Raised for HTTP-level failures during remote loading.

    Subclasses :class:`OSError` so that callers which already handle
    ``OSError`` (e.g. ``HDF5Video.__attrs_post_init__``) degrade gracefully.

    The original aiohttp/fsspec exception is intentionally *not* chained as
    ``__cause__`` / ``__context__`` because its string form embeds the raw URL
    (userinfo password, ``?token=`` query parameter, etc.), which
    :func:`traceback.format_exception` would otherwise leak. A redacted summary
    of the cause (type name + redacted message) is preserved instead via
    :attr:`cause_summary`.

    Attributes:
        url: Redacted URL (credentials stripped if present), or None.
        status: HTTP status code, or None for connection-level errors.
        cause_summary: Redacted, one-line summary of the underlying exception
            (type name plus redacted message), or None.
    """

    def __init__(
        self,
        message: str,
        *,
        url: str | None = None,
        status: int | None = None,
        cause_summary: str | None = None,
    ) -> None:
        """Build a RemoteIOError with a redacted, composed message.

        Args:
            message: Human-readable description of the failure.
            url: Raw URL associated with the failure. It is redacted before
                being stored or surfaced in the message.
            status: HTTP status code, if known.
            cause_summary: A pre-redacted summary of the underlying exception
                (see :func:`_redacted_cause_summary`). It is stored on the
                instance and appended to the message. The raw exception is
                never chained, so no credential-bearing string can leak through
                ``__cause__`` / ``__context__``.
        """
        self.url = _redact_url(url) if url else None
        self.status = status
        self.cause_summary = cause_summary
        parts = [message]
        if self.status is not None:
            parts.append(f"status={self.status}")
        if self.url:
            parts.append(f"url={self.url}")
        if self.cause_summary:
            parts.append(f"cause={self.cause_summary}")
        super().__init__("; ".join(parts))

doc = 'Raised for HTTP-level failures during remote loading.\n\nSubclasses :class:`OSError` so that callers which already handle\n``OSError`` (e.g. ``HDF5Video.__attrs_post_init``) degrade gracefully.\n\nThe original aiohttp/fsspec exception is intentionally not chained as\n``cause`` / ``context__`` because its string form embeds the raw URL\n(userinfo password, ``?token=`` query parameter, etc.), which\n:func:`traceback.format_exception` would otherwise leak. A redacted summary\nof the cause (type name + redacted message) is preserved instead via\n:attr:`cause_summary`.\n\nAttributes:\n url: Redacted URL (credentials stripped if present), or None.\n status: HTTP status code, or None for connection-level errors.\n cause_summary: Redacted, one-line summary of the underlying exception\n (type name plus redacted message), or None.\n' `class-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

`firstlineno = 113` `class-attribute` ¶

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.int(). For floating-point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer iteral.

int('0b100', base=0) 4

`module = 'sleap_io.io._remote'` `class-attribute` ¶

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.str() (if defined) or repr(object). encoding defaults to 'utf-8'. errors defaults to 'strict'.

`__static_attributes__ = ('cause_summary', 'status', 'url')` `class-attribute` ¶

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable's items.

If the argument is a tuple, the return value is the same object.

`weakref` `property` ¶

list of weak references to the object

`init(message, *, url=None, status=None, cause_summary=None)` ¶

Build a RemoteIOError with a redacted, composed message.

Parameters:

Name	Type	Description	Default
`message`	`str`	Human-readable description of the failure.	required
`url`	`str \| None`	Raw URL associated with the failure. It is redacted before being stored or surfaced in the message.	`None`
`status`	`int \| None`	HTTP status code, if known.	`None`
`cause_summary`	`str \| None`	A pre-redacted summary of the underlying exception (see :func:`_redacted_cause_summary`). It is stored on the instance and appended to the message. The raw exception is never chained, so no credential-bearing string can leak through `__cause__` / `__context__`.	`None`

Source code in sleap_io/io/_remote.py

def __init__(
    self,
    message: str,
    *,
    url: str | None = None,
    status: int | None = None,
    cause_summary: str | None = None,
) -> None:
    """Build a RemoteIOError with a redacted, composed message.

    Args:
        message: Human-readable description of the failure.
        url: Raw URL associated with the failure. It is redacted before
            being stored or surfaced in the message.
        status: HTTP status code, if known.
        cause_summary: A pre-redacted summary of the underlying exception
            (see :func:`_redacted_cause_summary`). It is stored on the
            instance and appended to the message. The raw exception is
            never chained, so no credential-bearing string can leak through
            ``__cause__`` / ``__context__``.
    """
    self.url = _redact_url(url) if url else None
    self.status = status
    self.cause_summary = cause_summary
    parts = [message]
    if self.status is not None:
        parts.append(f"status={self.status}")
    if self.url:
        parts.append(f"url={self.url}")
    if self.cause_summary:
        parts.append(f"cause={self.cause_summary}")
    super().__init__("; ".join(parts))

`clear_remote_cache(*, older_than=None, cache_storage=None)` ¶

Clear sleap-io's fsspec remote cache.

Only deletes files whose names match fsspec's cache-key pattern (sha-style hex hashes, optionally with a .tags sidecar), preventing accidental deletion of unrelated files if cache_storage points at a shared dir.

Parameters:

Name	Type	Description	Default
`older_than`	`float \| None`	If set, only delete files whose modification time is older than this many seconds.	`None`
`cache_storage`	`str \| PathLike \| None`	The cache directory to clear. Required to contain the sleap-io marker file. fsspec's built-in default cache directory is a per-process temporary directory (the `"TMP"` sentinel), which is not a stable, clearable location, so an explicit path is required here.	`None`

Returns:

Type	Description
`int`	The number of files deleted.

Raises:

Type	Description
`RuntimeError`	If `cache_storage` is None, a forbidden path (root, `$HOME`), or does not contain the sleap-io cache marker file.

Source code in sleap_io/io/_remote.py

def clear_remote_cache(
    *,
    older_than: float | None = None,
    cache_storage: str | os.PathLike | None = None,
) -> int:
    """Clear sleap-io's fsspec remote cache.

    Only deletes files whose names match fsspec's cache-key pattern (sha-style
    hex hashes, optionally with a ``.tags`` sidecar), preventing accidental
    deletion of unrelated files if ``cache_storage`` points at a shared dir.

    Args:
        older_than: If set, only delete files whose modification time is older
            than this many seconds.
        cache_storage: The cache directory to clear. Required to contain the
            sleap-io marker file. fsspec's built-in default cache directory is a
            per-process temporary directory (the ``"TMP"`` sentinel), which is
            not a stable, clearable location, so an explicit path is required
            here.

    Returns:
        The number of files deleted.

    Raises:
        RuntimeError: If ``cache_storage`` is None, a forbidden path (root,
            ``$HOME``), or does not contain the sleap-io cache marker file.
    """
    if cache_storage is None:
        raise RuntimeError(
            "clear_remote_cache requires an explicit cache_storage path: "
            "the same path passed to load_slp(..., cache_storage=...). fsspec's "
            "default cache directory is a per-process temporary directory and "
            "cannot be cleared reliably."
        )
    cache_dir = pathlib.Path(cache_storage).expanduser().resolve()

    forbidden = {pathlib.Path("/").resolve(), pathlib.Path.home().resolve()}
    if cache_dir in forbidden:
        raise RuntimeError(
            f"Refusing to clear cache: cache_storage={str(cache_dir)!r} is a "
            "forbidden path (root, $HOME, etc.)."
        )

    marker = cache_dir / _CACHE_MARKER_NAME
    if not marker.exists():
        raise RuntimeError(
            f"Refusing to clear cache: {str(cache_dir)!r} does not contain a "
            f"sleap-io cache marker file ({_CACHE_MARKER_NAME!r}). It was not "
            "written by sleap-io."
        )

    deleted = 0
    now = time.time()
    for p in cache_dir.iterdir():
        if not p.is_file():
            continue
        if not _CACHE_KEY_PATTERN.match(p.name):
            continue
        if older_than is not None:
            if (now - p.stat().st_mtime) < older_than:
                continue
        p.unlink()
        deleted += 1

    return deleted

`download(url, dest=None, *, headers=None, overwrite=False, progress=True, retries=3)` ¶

Download a remote file to local disk.

A simple, protocol-agnostic replacement for curl/wget in notebooks and demos. Supports the same URL schemes as :func:~sleap_io.load_file: http/https, cloud storage (s3/gs/gcs/az/abfs -- requires pip install 'sleap-io[cloud]'), and Google Drive share links. Unlike the streaming :func:~sleap_io.load_slp path, this writes the bytes to a local file (streamed to disk for HTTP/cloud, buffered in memory for Drive) and returns the path.

Parameters:

Name	Type	Description	Default
`url`	`str`	The remote URL to download.	required
`dest`	`str \| PathLike \| None`	Where to write the file. If `None` (default), the file is written to the current directory using the filename from the URL. If `dest` is an existing directory (or a string ending in a path separator), the file is written inside it using the URL filename. Otherwise `dest` is treated as the exact output path. Parent directories are created as needed.	`None`
`headers`	`dict[str, str] \| None`	Optional HTTP headers (HTTP/HTTPS only), e.g. `{"Authorization": "Bearer <token>"}`.	`None`
`overwrite`	`bool`	If False (default) and the destination already exists, the existing file is returned without re-downloading (idempotent, so re-running a notebook cell does not re-fetch large files). Set True to force a fresh download.	`False`
`progress`	`bool`	If True (default), show a `tqdm` progress bar (HTTP/cloud only; Drive downloads are buffered in memory without a byte bar).	`True`
`retries`	`int`	Maximum retries for transient HTTP errors (429/500/502/503/504), with exponential backoff honoring `Retry-After`. Applies to HTTP/cloud only; Google Drive files are fetched in a single buffered request and are not retried. Default: 3.	`3`

Returns:

Type	Description
`Path`	The local :class:`~pathlib.Path` of the downloaded (or already-present) file, resolved to an absolute path.

Raises:

Type	Description
`ValueError`	If `url` is not a remote URL, or no destination filename could be determined (the URL has no basename and `dest` does not name a file).
`IsADirectoryError`	If the destination resolves to an existing directory.
`RemoteIOError`	For HTTP / connection failures.
`ImportError`	For cloud schemes when the `[cloud]` extra is not installed.

Examples:

Download to the current directory (filename from the URL)::

import sleap_io as sio

path = sio.download("https://example.com/labels.slp")  # ./labels.slp

Download into a directory, or to an exact path::

sio.download("s3://bucket/run/video.mp4", "data/")  # data/video.mp4
sio.download("https://example.com/a.slp", "downloads/b.slp")

Fetch then load::

labels = sio.load_slp(sio.download(url))

Source code in sleap_io/io/_remote.py

def download(
    url: str,
    dest: str | os.PathLike | None = None,
    *,
    headers: dict[str, str] | None = None,
    overwrite: bool = False,
    progress: bool = True,
    retries: int = 3,
) -> pathlib.Path:
    """Download a remote file to local disk.

    A simple, protocol-agnostic replacement for ``curl``/``wget`` in notebooks
    and demos. Supports the same URL schemes as :func:`~sleap_io.load_file`:
    ``http``/``https``, cloud storage (``s3``/``gs``/``gcs``/``az``/``abfs`` --
    requires ``pip install 'sleap-io[cloud]'``), and Google Drive share links.
    Unlike the streaming :func:`~sleap_io.load_slp` path, this writes the bytes
    to a local file (streamed to disk for HTTP/cloud, buffered in memory for
    Drive) and returns the path.

    Args:
        url: The remote URL to download.
        dest: Where to write the file. If ``None`` (default), the file is written
            to the current directory using the filename from the URL. If ``dest``
            is an existing directory (or a string ending in a path separator), the
            file is written inside it using the URL filename. Otherwise ``dest``
            is treated as the exact output path. Parent directories are created as
            needed.
        headers: Optional HTTP headers (HTTP/HTTPS only), e.g.
            ``{"Authorization": "Bearer <token>"}``.
        overwrite: If False (default) and the destination already exists, the
            existing file is returned without re-downloading (idempotent, so
            re-running a notebook cell does not re-fetch large files). Set True to
            force a fresh download.
        progress: If True (default), show a ``tqdm`` progress bar (HTTP/cloud
            only; Drive downloads are buffered in memory without a byte bar).
        retries: Maximum retries for transient HTTP errors (429/500/502/503/504),
            with exponential backoff honoring ``Retry-After``. Applies to
            HTTP/cloud only; Google Drive files are fetched in a single buffered
            request and are not retried. Default: 3.

    Returns:
        The local :class:`~pathlib.Path` of the downloaded (or already-present)
        file, resolved to an absolute path.

    Raises:
        ValueError: If ``url`` is not a remote URL, or no destination filename
            could be determined (the URL has no basename and ``dest`` does not
            name a file).
        IsADirectoryError: If the destination resolves to an existing directory.
        RemoteIOError: For HTTP / connection failures.
        ImportError: For cloud schemes when the ``[cloud]`` extra is not
            installed.

    Examples:
        Download to the current directory (filename from the URL)::

            import sleap_io as sio

            path = sio.download("https://example.com/labels.slp")  # ./labels.slp

        Download into a directory, or to an exact path::

            sio.download("s3://bucket/run/video.mp4", "data/")  # data/video.mp4
            sio.download("https://example.com/a.slp", "downloads/b.slp")

        Fetch then load::

            labels = sio.load_slp(sio.download(url))
    """
    if not _is_url(url):
        raise ValueError(
            "download() expects a remote URL (http/https/s3/gs/gcs/az/abfs or a "
            f"Google Drive link), got: {_redact_url(os.fspath(url))!r}. There is "
            "nothing to download for a local path or unsupported scheme."
        )

    is_gdrive = _is_gdrive_url(url)

    # For HTTP/cloud the filename is known up front (URL basename), so an existing
    # destination can be returned without any network access. Google Drive URLs
    # carry no filename, so the target may be unknown until the bytes (and their
    # Content-Disposition) are fetched -- defer those to _download_gdrive.
    filename = None if is_gdrive else _filename_from_url(url)
    target = _resolve_target_path(dest, filename)

    if target is not None and target.is_dir():
        raise IsADirectoryError(
            f"Destination resolves to an existing directory: {target} (the URL "
            "basename collides with a subdirectory); pass an explicit file path "
            "as `dest`."
        )

    if target is not None and target.exists() and not overwrite:
        return target.resolve()

    if is_gdrive:
        return _download_gdrive(url, dest=dest, headers=headers, overwrite=overwrite)

    if target is None:
        raise ValueError(
            "Could not determine a destination filename from the URL "
            f"({_redact_url(url)}); pass an explicit file path as `dest`."
        )

    return _download_fsspec(
        url,
        target,
        headers=headers,
        progress=progress,
        retries=retries,
    )

`open_remote_h5(url, *, headers=None)` ¶

Open a remote HDF5 file for membership/existence probing.

Thin convenience wrapper around :func:open_url using blockcache mode, used by Video.exists to probe a remote .slp/pkg.slp URL.

Parameters:

Name	Type	Description	Default
`url`	`str`	The remote HDF5 URL.	required
`headers`	`dict[str, str] \| None`	Optional HTTP headers.	`None`

Returns:

Type	Description
	A file-like object suitable for `h5py.File(...)`.

Source code in sleap_io/io/_remote.py

def open_remote_h5(url: str, *, headers: dict[str, str] | None = None):
    """Open a remote HDF5 file for membership/existence probing.

    Thin convenience wrapper around :func:`open_url` using ``blockcache`` mode,
    used by ``Video.exists`` to probe a remote ``.slp``/``pkg.slp`` URL.

    Args:
        url: The remote HDF5 URL.
        headers: Optional HTTP headers.

    Returns:
        A file-like object suitable for ``h5py.File(...)``.
    """
    return open_url(url, headers=headers, stream_mode="blockcache")

`open_url(url, *, headers=None, stream_mode='auto', cache_storage=None, cache_expiry=None, block_size=1048576, max_blocks=32, retries=3)` ¶

Open url as a file-like object using the configured strategy.

Parameters:

Name	Type	Description	Default
`url`	`str`	The remote URL to open.	required
`headers`	`dict[str, str] \| None`	HTTP headers (HTTP/HTTPS only).	`None`
`stream_mode`	`str`	One of `"auto"` (alias for `"blockcache"`), `"blockcache"`, `"cache"` (simplecache), `"filecache"`, or `"download"` (ephemeral full read into BytesIO).	`'auto'`
`cache_storage`	`str \| PathLike \| None`	Override the cache directory for cache/filecache modes.	`None`
`cache_expiry`	`float \| None`	TTL (seconds) for `filecache` revalidation. Defaults to 3600 when not given.	`None`
`block_size`	`int`	Range block size in bytes for `blockcache`.	`1048576`
`max_blocks`	`int`	Max in-memory LRU blocks per open file for `blockcache`.	`32`
`retries`	`int`	Maximum retries for transient HTTP errors (429/500/502/503/504), with exponential backoff honoring `Retry-After`. Default: 3.	`3`

Returns:

Type	Description
	A file-like object (fsspec buffered file or `io.BytesIO`) ready to be passed to `h5py.File(...)` or a reader.

Raises:

Type	Description
`RemoteIOError`	For HTTP / connection failures.
`ImportError`	For cloud schemes when the extra is not installed.
`ValueError`	For an unrecognized `stream_mode`.

Source code in sleap_io/io/_remote.py

def open_url(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    stream_mode: str = "auto",
    cache_storage: str | os.PathLike | None = None,
    cache_expiry: float | None = None,
    block_size: int = 1 << 20,
    max_blocks: int = 32,
    retries: int = 3,
):
    """Open ``url`` as a file-like object using the configured strategy.

    Args:
        url: The remote URL to open.
        headers: HTTP headers (HTTP/HTTPS only).
        stream_mode: One of ``"auto"`` (alias for ``"blockcache"``),
            ``"blockcache"``, ``"cache"`` (simplecache), ``"filecache"``, or
            ``"download"`` (ephemeral full read into BytesIO).
        cache_storage: Override the cache directory for cache/filecache modes.
        cache_expiry: TTL (seconds) for ``filecache`` revalidation. Defaults to
            3600 when not given.
        block_size: Range block size in bytes for ``blockcache``.
        max_blocks: Max in-memory LRU blocks per open file for ``blockcache``.
        retries: Maximum retries for transient HTTP errors (429/500/502/503/504),
            with exponential backoff honoring ``Retry-After``. Default: 3.

    Returns:
        A file-like object (fsspec buffered file or ``io.BytesIO``) ready to be
        passed to ``h5py.File(...)`` or a reader.

    Raises:
        RemoteIOError: For HTTP / connection failures.
        ImportError: For cloud schemes when the extra is not installed.
        ValueError: For an unrecognized ``stream_mode``.
    """
    # Google Drive share links resolve through a two-hop interstitial flow and
    # are full-prefetched into memory (Drive's HEAD 405 + quota-mid-stream
    # behavior makes lazy range reads unreliable). The streaming kwargs above
    # (stream_mode/cache/block_size/...) do not apply to this branch. Detection
    # (``_is_gdrive_url``) is local + pure stdlib; the resolver is imported
    # lazily only when a Drive URL is actually seen.
    if _is_gdrive_url(url):
        from sleap_io.io._gdrive import _open_gdrive

        return _open_gdrive(url, headers=headers)

    parsed = urllib.parse.urlparse(url)
    scheme = parsed.scheme.lower()
    fs = _build_fsspec_filesystem(
        scheme, headers=headers, block_size=block_size, max_blocks=max_blocks
    )

    if stream_mode == "auto":
        stream_mode = "blockcache"

    if stream_mode not in ("blockcache", "cache", "filecache", "download"):
        raise ValueError(
            f"Invalid stream_mode={stream_mode!r}; expected one of "
            "{'auto', 'blockcache', 'cache', 'filecache', 'download'}."
        )

    if stream_mode == "blockcache":

        def _open():
            return fs.open(
                url,
                mode="rb",
                cache_type="blockcache",
                block_size=block_size,
                # fsspec's BlockCache uses ``maxblocks`` (no underscore).
                cache_options={"maxblocks": max_blocks},
            )

    elif stream_mode == "cache":

        def _open():
            import fsspec

            # ``skip_instance_cache`` on the outer cache filesystem too: the
            # chained ``fsspec.open`` builds (and caches) a SimpleCacheFileSystem
            # eagerly, so without this the outer cache grows per distinct header
            # set even though the inner http fs is already skipped.
            simplecache_opts: dict[str, Any] = {"skip_instance_cache": True}
            if cache_storage is not None:
                _mark_cache_dir(cache_storage)
                simplecache_opts["cache_storage"] = str(cache_storage)
            return fsspec.open(
                f"simplecache::{url}",
                mode="rb",
                simplecache=simplecache_opts,
                http=_http_inner_options(headers),
            ).open()

    elif stream_mode == "filecache":

        def _open():
            import fsspec

            options: dict[str, Any] = {
                "expiry_time": cache_expiry if cache_expiry is not None else 3600,
                # See the ``cache`` branch above: skip fsspec's instance cache on
                # the outer WholeFileCacheFileSystem so it does not grow per
                # distinct header set in a long-lived process.
                "skip_instance_cache": True,
            }
            if cache_storage is not None:
                _mark_cache_dir(cache_storage)
                options["cache_storage"] = str(cache_storage)
            return fsspec.open(
                f"filecache::{url}",
                mode="rb",
                filecache=options,
                http=_http_inner_options(headers),
            ).open()

    else:  # stream_mode == "download"

        def _open():
            with fs.open(url, mode="rb") as src:
                return io.BytesIO(src.read())

    return _open_with_retries(_open, url=url, retries=retries)

_remote

sleap_io.io._remote ¶

__annotations__ = {'_FORMAT_MAGIC': 'tuple[tuple[bytes, str], ...]'} module-attribute ¶

__cached__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/__pycache__/_remote.cpython-313.pyc' module-attribute ¶

__file__ = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/_remote.py' module-attribute ¶

__name__ = 'sleap_io.io._remote' module-attribute ¶

__package__ = 'sleap_io.io' module-attribute ¶

RemoteIOError ¶

__firstlineno__ = 113 class-attribute ¶

__module__ = 'sleap_io.io._remote' class-attribute ¶

__static_attributes__ = ('cause_summary', 'status', 'url') class-attribute ¶

__weakref__ property ¶

__init__(message, *, url=None, status=None, cause_summary=None) ¶

clear_remote_cache(*, older_than=None, cache_storage=None) ¶

download(url, dest=None, *, headers=None, overwrite=False, progress=True, retries=3) ¶

open_remote_h5(url, *, headers=None) ¶

open_url(url, *, headers=None, stream_mode='auto', cache_storage=None, cache_expiry=None, block_size=1048576, max_blocks=32, retries=3) ¶

`sleap_io.io._remote` ¶

`annotations = {'_FORMAT_MAGIC': 'tuple[tuple[bytes, str], ...]'}` `module-attribute` ¶

`cached = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/pycache/_remote.cpython-313.pyc'` `module-attribute` ¶

`file = '/home/runner/work/sleap-io/sleap-io/sleap_io/io/_remote.py'` `module-attribute` ¶

`name = 'sleap_io.io._remote'` `module-attribute` ¶

`package = 'sleap_io.io'` `module-attribute` ¶

`RemoteIOError` ¶

`firstlineno = 113` `class-attribute` ¶

`module = 'sleap_io.io._remote'` `class-attribute` ¶

`__static_attributes__ = ('cause_summary', 'status', 'url')` `class-attribute` ¶

`weakref` `property` ¶

`init(message, *, url=None, status=None, cause_summary=None)` ¶

`clear_remote_cache(*, older_than=None, cache_storage=None)` ¶

`download(url, dest=None, *, headers=None, overwrite=False, progress=True, retries=3)` ¶

`open_remote_h5(url, *, headers=None)` ¶

`open_url(url, *, headers=None, stream_mode='auto', cache_storage=None, cache_expiry=None, block_size=1048576, max_blocks=32, retries=3)` ¶