Skip to content

MEDS_extract.download

The shared download layer for MEDS_extract-based ETLs: a small, transport-agnostic API for staging a dataset’s raw files into a local directory before the MEDS_extract stage pipeline runs.

Why this submodule exists

A MEDS ETL can’t start until the raw source files are sitting on local disk. Getting them there is deceptively fiddly — datasets live behind credentialed HTTP endpoints, PhysioNet release manifests, S3/GCS buckets, or a colleague’s pre-downloaded mirror; they need checksum verification, resumable transfers, and politeness toward rate-limited hosts.

This submodule exists so that a downstream ETL never has to write download code at all. Instead, it declares where its raw files live in a standardized sources: block in its MESSY spec, and meds-extract-download turns that declaration into a deterministic, verifiable local copy. The goals:

  • One specification structure. Every ETL describes its raw data the same way — a sources: block of typed backend entries — so the “how do I get this dataset” question has a uniform, reviewable answer that lives next to the rest of the spec.
  • One toolchain. A single CLI (meds-extract-download) and a single Python API (Source.download_all) stage any dataset, regardless of where it’s hosted. ETL authors compose backends; they don’t reimplement transports.
  • Deterministic, verified retrieval. SHA-256 verification, atomic writes, and a verify-or-re-fetch policy mean a completed download contains exactly the manifest’s files: local copies that verify are skipped, anything that can’t be verified is re-fetched — no silently-stale local copies leaking into a pipeline run.

The submodule sits alongside the MEDS-transforms stage DAG, not inside it: download I/O is network / blob storage rather than sharded parquet, parallelism is per-file transport streams rather than per-shard workers, and failures are partial-retry / resume rather than per-stage. It keeps the same ergonomics as a stage (Hydra-driven, CLI-addressable, override-friendly) — just with its own machinery.

Overview

At the highest level, staging a dataset is four steps:

  1. A MESSY spec declares where its raw files live in a sources: block.
  2. spec.py (sources_from_spec) turns each entry into a Source instance — HTTPSource, FsspecSource, or PhysioNetSource.
  3. Source.download_all is called on each source…
  4. …staging every file into one shared output_dir/.

A Source is anywhere raw data comes from. It knows two things: what files it offers (_list_files) and how to stream one file’s bytes to a local path (_pull). Everything else — .part staging, SHA-256 verification, atomic rename, the skip/re-fetch policy, manifest validation and include/exclude filtering, duplicate-destination detection, sequential-vs-parallel orchestration, error aggregation — lives once on the Source ABC and is shared by every backend.

Using it from the CLI

The common case. A MESSY spec declares its backends:

sources:
  dataset: # the bucket selected by `key=` (default: "dataset")
    - type: physionet
      base_url: https://physionet.org/files/mimiciv/3.1
      username: ${oc.env:PHYSIONET_USER}
      password: ${oc.env:PHYSIONET_PASS}
      include: # optional fnmatch globs — stage only what the ETL reads
        - hosp/*.csv.gz
        - icu/*.csv.gz
  common: # always appended, regardless of `key=`
    - type: http
      urls:
        - https://raw.githubusercontent.com/.../concept_map.csv

Basic auth alone is not enough for PhysioNet: physionet.org serves credentialed /files/ paths only to clients whose User-Agent starts with Wget/<version> (a prefix match — anything appended after is preserved), and rejects other UAs with a 403 before credentials are considered. PhysioNetSource therefore defaults its client’s UA to Wget/<version> MEDS-Extract/<version> — passing the gate while staying honestly identified. A headers: {User-Agent: ...} entry in the source config overrides it completely.

and meds-extract-download stages it (Hydra dotlist overrides, one command):

meds-extract-download spec=/path/to/messy.yaml output_dir=/path/to/raw key=dataset concurrency=4

The override knobs:

  • spec — the MESSY spec file. Besides a filesystem path, pkg:// syntax reaches a spec bundled inside an installed package (e.g. spec=pkg://MIMIC_IV_MEDS.configs.event_configs.yaml) — resolved via MEDS-transforms’ resolve_pkg_path, the same syntax MEDS_transform-pipeline accepts for pipeline configs.
  • key — which sources: bucket to pull; common is always appended. When the spec declares sources buckets, a key naming none of them is an error, not a silent no-op (a spec with no sources: block at all warns and exits 0 — a legitimately download-free ETL). The reserved dataset_version key (raw-data version metadata — scalar string or {bucket: version} mapping, interpolatable from bucket entries via ${sources.dataset_version}; consumed by meds-extract-run for version stamping) is never a bucket and cannot be selected.
  • concurrency — size of the one thread pool shared across all sources.
  • continue_on_error — collect per-file failures and keep going (all sources are attempted; the process exits non-zero at the end if anything failed). With the default False, the first failing source stops the whole run.
  • do_overwrite — re-fetch every file even if a verified local copy exists.

Before any fetch, the CLI materializes every source’s manifest and rejects cross-source destination collisions (two sources listing the same rel_path), so a misconfigured spec fails precisely and immediately rather than mid-download. The process exits 0 only on full success.

Using it from Python

The CLI is a thin wrapper over the library API. download_all is the single entry point — sequential by default, parallel when handed a pool. The example below is a runnable doctest (a local fsspec source standing in for a remote host):

>>> from concurrent.futures import ThreadPoolExecutor
>>> from MEDS_extract.download import FsspecSource, validate_unique_destinations
>>> mirror = '''
... patients.csv: "patient_id,dob\\n1,2000-01-01\\n"
... labs:
...   vitals.csv: "pid,hr\\n1,80\\n"
... '''
>>> with yaml_disk(mirror) as src_dir, tempfile.TemporaryDirectory() as raw:
...     sources = [FsspecSource(root=str(src_dir))]  # what sources_from_spec returns
...     validate_unique_destinations(sources)  # reject cross-source collisions up-front
...     with ThreadPoolExecutor(max_workers=4) as pool:
...         for src in sources:
...             with src:  # close() owned network clients on exit
...                 src.download_all(raw, pool=pool)
...     print_directory(Path(raw))
├── labs
│   └── vitals.csv
└── patients.csv

For a single source the whole dance collapses to FsspecSource(root=...).download_all("raw/") — sequential, no pool, nothing to close for backends that own no network client (HTTP-backed sources should use with).

The rest of this document walks through the pieces behind that API.

Files

File Responsibility
source.py The Source ABC, the RemoteFile manifest row, ChecksumError, sha256_of, validate_unique_destinations, and the whole orchestration loop (download_all + helpers).
backends/http.py HTTPSource — explicit list of URLs. tenacity-retried manifest GETs + streaming, .part-file Range-resume download, Content-Range validation. No crawling.
backends/physionet.py PhysioNetSource(HTTPSource) — discovers its file list from the SHA256SUMS.txt manifest every PhysioNet release publishes. Overrides _list_files (plus its constructor). Defaults the client’s User-Agent to a Wget/<version>-prefixed string (physionet’s /files/ gate) and turns the gate’s challenge-less 403 into a legible error.
backends/fsspec.py FsspecSource — any fsspec protocol via universal_pathlib (file://, s3://, gs://, …). For re-runs against a pre-downloaded local / cloud mirror.
unarchive.py ArchiveFormat + safe_extract — post-fetch archive extraction (zip / tar / tar.gz) with zip-slip / tar-slip validation before any bytes are written. Opt-in via RemoteFile.unarchive.
spec.py source_from_config / sources_from_spec — turn raw sources: YAML entries into concrete Source instances. The one place the type: → class registry lives.
cli.py meds-extract-download — the Hydra entry point. Resolves the spec, builds + cross-validates the sources, owns the shared thread pool, drives every source, exits non-zero on failure.
backends/__init__.py Lazily re-exports the three backend classes (PEP 562), so the HTTP stack is only imported when actually used.
__init__.py Public surface: Source, RemoteFile, ChecksumError, the three backends, source_from_config, sources_from_spec, validate_unique_destinations.

Architecture

Source — the ABC

Concrete backends implement exactly two hooks:

@abstractmethod
def _list_files(self) -> Iterable[RemoteFile]: ...  # enumerate files
@abstractmethod
def _pull(self, source_path: str, target: Path) -> None: ...  # stream bytes

The base class supplies everything else. _fetch_one(item, dest_dir, do_overwrite) is the per-file pipeline (orchestrator-facing): resolve dest, apply the skip/re-fetch policy, derive .part, call _pull, verify SHA-256, atomic-rename.

The user-facing entry points:

  • download_all(dest_dir, *, pool=None, continue_on_error=False, do_overwrite=False) — the single public fetch entry point.
  • files — a cached_property wrapping _list_files(): materializes the manifest to a list, applies the constructor’s include / exclude globs, validates that every rel_path is unique, and caches the result so a network-backed manifest (PhysioNet’s SHA256SUMS.txt) isn’t re-fetched on a second download_all call. n_files is the corresponding count.
  • close() / __enter__ / __exit__ — resource lifecycle. Backends that own a network client (HTTPSource) override close(); the CLI registers every source with an ExitStack so clients are released deterministically.

Every backend constructor also accepts include / exclude — fnmatch-style globs over the (normalized) rel_path — so an ETL can stage only the subset of a large release it actually reads (e.g. include: ["hosp/*.csv.gz"] against a release that also bundles waveforms or images).

RemoteFile — the manifest row

A frozen, self-validating dataclass. Every _list_files implementation — in-repo backend or downstream Source subclass — constructs these, so it is exported from MEDS_extract.download:

@dataclass(frozen=True)
class RemoteFile:
    rel_path: str  # where it lands under dest_dir (forward slashes)
    source_path: str  # transport's source-side address (URL / UPath spec)
    sha256: str | None = None  # the only verifier the orchestrator trusts
    unarchive: str | None = None  # post-fetch unpack: zip / tar / tar.gz / tgz / auto
    cleanup_archive: bool | None = None  # tri-state: None defers to the unarchive mode

Validation runs in __post_init__, so a malformed row fails the instant it is built: rel_path must be relative, forward-slash, and must not escape the destination directory after normalization; sha256 (when set) must be 64 hex chars and is normalized to lowercase. The cross-row check — no two rows resolving to the same destination — lives in Source.files, and the cross-source variant in validate_unique_destinations.

sha256 is the only verifier the orchestrator trusts to skip a re-fetch. A RemoteFile with no sha256 can still be downloaded, but on a re-run it can’t be skipped — it is re-fetched; see the skip/re-fetch policy below.

The orchestration loop

download_all is one straight pass: get the validated manifest, turn it into a stream of fetch attempts, and run them through a single error-collection loop.

  1. self.files — the validated, filtered, cached manifest.
  2. _iter_attempts(...) pairs each row with the zero-arg thunk that fetches it; _attempts(...) dispatches those pairs by mode:
    • no pool → the pairs pass through; the callable runs _fetch_one in the calling thread when invoked.
    • pool given → every thunk is submitted up front; pairs come back as (item, future.result) in completion order.
  3. A single for item, run in attempts: loop calls run(). On a per-file exception: with continue_on_error=True the error is collected and the loop continues; otherwise it propagates immediately (and in pooled mode the still-queued futures are cancelled on the way out).
  4. If any errors were collected, they are raised together as one ExceptionGroup; otherwise download_all returns None.

_fetch_one is where the per-file skip / re-fetch policy lives:

dest state do_overwrite=False do_overwrite=True
doesn’t exist fetch fetch
exists, verifies against manifest sha256 skip clear + refetch
exists, sha mismatch refetch (with a warning) clear + refetch
exists, no manifest sha refetch clear + refetch

The “exists but can’t verify → re-fetch” rule is intentional: a file we can’t prove matches the manifest is never trusted (silently skipping it is how stale or half-flushed local copies leak into a pipeline run), and never skipped. It keeps re-runs of mixed manifests — sha-verified PhysioNet files alongside checksum-free HTTP URLs — cheap and idempotent: verified files skip, everything else re-fetches. The re-fetch still stages into .part and replaces dest only via the atomic rename (after the fresh bytes verify, when a sha exists), so a failed re-fetch never destroys the existing copy. do_overwrite=True forces a clean re-fetch of everything, verified or not.

Two .part-level refinements: a leftover .part that already verifies against the manifest sha is promoted to dest directly (a prior run died between the last byte and the rename — no re-fetch needed), and a leftover .part with no manifest sha to verify against is discarded (resume-without-verification is unsafe).

Post-fetch unarchive (opt-in)

Some releases ship their data as a single archive the pipeline can’t read directly (polars reads .csv.gz natively — but not members inside a .zip / .tar.gz). Setting unarchive: on a RemoteFile — per URL entry on HTTPSource, or source-wide on PhysioNetSource — makes _fetch_one unpack the archive into the dest’s directory right after the atomic rename ("fetched" and "promoted" paths only; a "skipped" dest is not re-extracted). Extraction happens after SHA-256 verification, so the hash always describes the archive as transferred, never the extracted tree. safe_extract validates every member (absolute paths, .., symlink/hardlink targets) before writing any bytes, and tar extraction additionally applies PEP 706’s data_filter. cleanup_archive: is tri-state: None defers to the mode (auto drops the archive after extraction, explicit formats keep it); True / False always wins. Note that dropping the archive also drops the skip-on-rerun evidence — the next download_all will re-fetch it.

Pool ownership

download_all runs sequentially by default; pass an Executor (typically a ThreadPoolExecutor) to opt into parallelism, and the caller owns its lifetime. That gives:

  • Caller-controlled worker cap — sized to whatever the transport tolerates (one rate-limited host vs. ten fast ones is a per-deployment decision).
  • One pool shared across sources — the CLI builds a single pool sized to concurrency= and hands it to every source’s download_all, so the bound is global.
  • Deterministic teardown — the CLI shuts the pool down with shutdown(wait=False, cancel_futures=True), so a Ctrl+C mid-download cancels all queued work immediately. Worker threads are not daemons (Python ≥3.9 joins them at interpreter exit), so in-flight transfers end when their transport is torn down: the CLI closes each HTTP source’s client on the way out, which kills live streams promptly; fsspec copies have no abort path and run to completion.

spec.py — spec → objects

source_from_config({"type": "http", "urls": [...]}) looks the type: string up in the _SOURCE_TYPES registry — the one dict mapping each type to its backend module and class (the “supported types” error message derives from the same dict, so the two can’t drift). Backend modules import lazily, per selected type. sources_from_spec(spec, key="dataset") reads a whole sources: block, pulls the selected bucket plus the always-appended common: bucket, and returns the constructed list. New backend = new backends/ module + one registry row.

cli.py — the meds-extract-download entry point

A Hydra entry point (DownloadConfig is a hydra_registered_dataclass). It:

  1. resolves the spec path against the user’s original CWD (Hydra changes CWD);
  2. resolves OmegaConf interpolations on only the sources: subtree — so a combined MESSY file’s unrelated ${oc.env:...} interpolations in the event-conversion section don’t need to be set just to download;
  3. validates key= against the buckets the spec actually declares, then builds the sources via sources_from_spec;
  4. opens an ExitStack, creates one shared pool, registers every source for close(), rejects cross-source destination collisions via validate_unique_destinations, and calls download_all on each — stopping at the first failed source unless continue_on_error=true;
  5. exits 0 on full success and 1 otherwise (via explicit sys.exit — Hydra discards the task function’s return value, so returning an exit code would not work).

Adding a backend

  1. Add backends/<name>.py with a class FooSource(Source) implementing _list_files and _pull. The base class wraps _pull with .part staging, SHA-256 verification, and atomic rename — _pull’s only contract is “produce a complete file at target or raise.” Accept include / exclude in your constructor and forward them to super().__init__ so manifest filtering works uniformly.
  2. Add one row to _SOURCE_TYPES in spec.py.
  3. Cover it with doctests in the backend module (per the project’s doctest-first convention) and add wire-level tests to tests/test_download.py if it needs a real transport round-trip. Backends that work without the download extra should add their tests to tests/test_download_fsspec.py instead, so the no-extras CI job runs them.

Testing

  • Doctests in each module cover the pure logic: spec dispatch, URL normalization, RemoteFile validation, SHA256SUMS.txt parsing, manifest filtering, and the Source.download_all skip/re-fetch/traversal/dup paths (via stub sources in the source.py docstrings). This README’s Python-usage example is itself a collected doctest.
  • tests/test_download.py covers what doctests can’t: _resumable_stream’s wire-level behavior (Range resume, 416/206 mismatch handling, identity content-coding) against httpx.MockTransport, streaming retry behavior, the Source._fetch_one staging pipeline (sha verify + atomic rename + .part promotion/discard), end-to-end download_all flows (sequential and pooled), and the SIGINT-cancellation regression (which needs a real signal in a real subprocess — tests/_fetcher_sigint_child.py). Requires the download extra; the no-extras environment skips it.
  • tests/test_download_fsspec.py — the extras-free path: the meds-extract-download CLI subprocess flows (success, failure exit codes, key validation and bucket selection, cross-source collision rejection, fail-fast across sources, sources-subtree-only interpolation, the no-sources:-block warn-and-exit-0 contract) and FsspecSource behavior including a non-local (memory://) protocol and filter-before-hashing. Runs in the no-extras CI job.
  • tests/test_example.py exercises the real PhysioNet path end-to-end (gated behind the integration marker).

Shared download layer for MEDS_extract-based ETLs.

The public surface is:

  • :class:Source ABC — every backend implements this. Public entry point is :meth:Source.download_all.
  • :class:RemoteFile — the validated manifest row every _list_files implementation (in-repo backend or downstream :class:Source subclass) constructs.
  • :class:ChecksumError — raised by download_all (directly or wrapped in an ExceptionGroup) when a fetched file’s SHA-256 doesn’t match the manifest.
  • :func:validate_unique_destinations — cross-source collision check for callers staging several sources into one shared directory (the CLI runs it automatically).
  • Concrete backends: :class:HTTPSource, :class:FsspecSource, :class:PhysioNetSource
  • :func:sources_from_spec / :func:source_from_config — build :class:Source instances from a MESSY sources: block

See https://github.com/mmcdermott/MEDS_extract/issues/81 for the design rationale.

The heavy HTTP deps (:mod:httpx, :mod:tenacity) are declared under the download extra in pyproject.toml and imported lazily — only accessing :class:HTTPSource / :class:PhysioNetSource (directly or via a type: http / type: physionet spec entry) requires them. FsspecSource, the Source ABC, and the spec helpers work from the base install. Install the extra with pip install 'MEDS_extract[download]'.

ChecksumError

Bases: ValueError

Raised when a downloaded file’s SHA-256 doesn’t match the expected digest.

Every :class:Source that honors remote.sha256 raises this on mismatch (not just the HTTP-backed ones) so callers can catch a single exception type regardless of transport.

Source code in MEDS_extract/download/source.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
class ChecksumError(ValueError):
    """Raised when a downloaded file's SHA-256 doesn't match the expected digest.

    Every :class:`Source` that honors ``remote.sha256`` raises this on mismatch (not
    just the HTTP-backed ones) so callers can catch a single exception type
    regardless of transport.
    """

    def __init__(self, source_id: str, expected: str, actual: str):
        self.source_id = source_id
        self.expected = expected
        self.actual = actual
        super().__init__(f"SHA-256 mismatch for {source_id}: expected {expected}, got {actual}")

FsspecSource

Bases: Source

A :class:Source backed by an fsspec-compatible root via :class:upath.UPath.

Accepts any protocol UPath supports: file:// and local paths (re-run against a pre-downloaded copy), s3:// / gs:// / azure:// (mirrors the user keeps on cloud storage), and so on. The fsspec extras for cloud protocols (s3fs, gcsfs, …) are NOT declared as dependencies here — users install them themselves following the standard fsspec pattern.

Each :class:RemoteFile carries a SHA-256 computed from the source file at _list_files time, so re-runs verify the on-disk copy against the same hash and skip if it matches. Cost note: computing those hashes reads every selected source file in full, serially, in the manifest-building thread — before any file is fetched and on every run (the manifest is cached per Source instance, not across processes). For local roots that read is cheap; for cloud-bucket roots it means a complete remote read of the (filtered) dataset per invocation, so use include= / exclude= to subset large mirrors, or prefer a local mirror for iterated re-runs.

Parameters:

Name Type Description Default
root str

The tree to copy from — a local path or any UPath-supported URL. Must exist; a nonexistent root raises :class:FileNotFoundError at manifest time rather than silently yielding an empty file list.

 required
include, exclude

Optional :mod:fnmatch globs applied to the manifest — see :class:~MEDS_extract.download.source.Source.

 required

Examples:

download_all walks the tree and copies every file under dest_dir, preserving the relative layout:

>>> spec = '''
... patients.csv: |
...   patient_id,dob
...   1,2000-01-01
... labs:
...   vitals.csv: |
...     pid,hr
...     1,80
... '''
>>> with yaml_disk(spec) as src_dir, tempfile.TemporaryDirectory() as dst:
...     dst = Path(dst)
...     FsspecSource(root=str(src_dir)).download_all(dst)
...     print_directory(dst)
├── labs
│   └── vitals.csv
└── patients.csv

A nonexistent root is a config error, not an empty dataset:

>>> FsspecSource(root="/no/such/dir/anywhere").files
Traceback (most recent call last):
    ...
FileNotFoundError: FsspecSource root does not exist: /no/such/dir/anywhere
Source code in MEDS_extract/download/backends/fsspec.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
class FsspecSource(Source):
    """A :class:`Source` backed by an fsspec-compatible root via :class:`upath.UPath`.

    Accepts any protocol UPath supports: ``file://`` and local paths (re-run against a
    pre-downloaded copy), ``s3://`` / ``gs://`` / ``azure://`` (mirrors the user keeps on
    cloud storage), and so on. The ``fsspec`` extras for cloud protocols (``s3fs``,
    ``gcsfs``, …) are NOT declared as dependencies here — users install them themselves
    following the standard fsspec pattern.

    Each :class:`RemoteFile` carries a SHA-256 computed from the source file at
    ``_list_files`` time, so re-runs verify the on-disk copy against the same hash and
    skip if it matches. **Cost note**: computing those hashes reads every selected
    source file in full, serially, in the manifest-building thread — *before* any file
    is fetched and *on every run* (the manifest is cached per ``Source`` instance, not
    across processes). For local roots that read is cheap; for cloud-bucket roots it
    means a complete remote read of the (filtered) dataset per invocation, so use
    ``include=`` / ``exclude=`` to subset large mirrors, or prefer a local mirror for
    iterated re-runs.

    Args:
        root: The tree to copy from — a local path or any UPath-supported URL.
            Must exist; a nonexistent root raises :class:`FileNotFoundError` at
            manifest time rather than silently yielding an empty file list.
        include, exclude: Optional :mod:`fnmatch` globs applied to the manifest —
            see :class:`~MEDS_extract.download.source.Source`.

    Examples:
        ``download_all`` walks the tree and copies every file under ``dest_dir``,
        preserving the relative layout:

        >>> spec = '''
        ... patients.csv: |
        ...   patient_id,dob
        ...   1,2000-01-01
        ... labs:
        ...   vitals.csv: |
        ...     pid,hr
        ...     1,80
        ... '''
        >>> with yaml_disk(spec) as src_dir, tempfile.TemporaryDirectory() as dst:
        ...     dst = Path(dst)
        ...     FsspecSource(root=str(src_dir)).download_all(dst)
        ...     print_directory(dst)
        ├── labs
        │   └── vitals.csv
        └── patients.csv

        A nonexistent root is a config error, not an empty dataset:

        >>> FsspecSource(root="/no/such/dir/anywhere").files
        Traceback (most recent call last):
            ...
        FileNotFoundError: FsspecSource root does not exist: /no/such/dir/anywhere
    """

    def __init__(self, root: str, include: list[str] | None = None, exclude: list[str] | None = None):
        super().__init__(include=include, exclude=exclude)
        self._root = UPath(root)

    def _list_files(self) -> Iterable[RemoteFile]:
        if not self._root.exists():
            raise FileNotFoundError(f"{type(self).__name__} root does not exist: {self._root}")
        logger.info(
            f"Building manifest for {self._root}: hashing each selected source file "
            "(this reads the selected files in full)"
        )
        n_total = 0
        n_skipped = 0
        for p in self._root.rglob("*"):
            if not p.is_file():
                continue
            rel_path = p.relative_to(self._root).as_posix()
            n_total += 1
            # Apply the manifest filters *before* hashing — the whole point of
            # ``include=`` on a cloud mirror is to not read the excluded bytes.
            if not self._selected_path(posixpath.normpath(rel_path)):
                n_skipped += 1
                continue
            logger.debug(f"Hashing {rel_path}")
            yield RemoteFile(
                rel_path=rel_path,
                sha256=sha256_of(p),
                source_path=str(p),
            )
        if n_skipped:
            logger.info(
                f"include/exclude filters skipped {n_skipped}/{n_total} files under "
                f"{self._root} before hashing"
            )

    def _pull(self, source_path: str, target: Path) -> None:
        with UPath(source_path).open("rb") as src, target.open("wb") as dst:
            shutil.copyfileobj(src, dst, length=1024 * 1024)

HTTPSource

Bases: Source

A :class:Source backed by an explicit list of HTTP URLs.

Use this for shared metadata downloads where the file list is known up-front — e.g. MIMIC’s common: block of concept-map CSVs from raw.githubusercontent.com. No crawling, no manifest parsing.

Each URL entry can be either a plain string or a dict with optional sha256 and rel_path fields. rel_path defaults to the URL’s basename.

Parameters:

Name Type Description Default
urls list[str | dict] | None

List of URL entries — plain strings or dicts. Subclasses that discover URLs at :meth:_list_files time (e.g. :class:PhysioNetSource) may pass None.

 None
client Client | None

Optional pre-built :class:httpx.Client. When omitted, one is built via :meth:_make_client with the remaining kwargs.

 None
auth, headers, timeout, transport

Forwarded to :meth:_make_client when client is not provided. headers is a {name: value} mapping applied as default headers on every request — used for API-key auth (X-Dataverse-key, bearer tokens) and content negotiation (Accept:).

 required
max_attempts, retry_wait

Govern the shared retry policy (:meth:_retrying) applied to both manifest GETs (:meth:_get) and streaming downloads (:meth:_pull) — regardless of whether client was injected.

 required
include, exclude

Optional :mod:fnmatch globs applied to the manifest — see :class:~MEDS_extract.download.source.Source.

 required

Examples:

Plain string URLs resolve to basename-based relative paths:

>>> src = HTTPSource(urls=["https://example.com/foo.csv", "https://example.com/bar.csv"])
>>> [r.rel_path for r in src._list_files()]
['foo.csv', 'bar.csv']
>>> src.close()

Dict entries can override rel_path and provide a checksum:

>>> src = HTTPSource(
...     urls=[
...         {"url": "https://example.com/foo.csv", "rel_path": "lookups/foo.csv"},
...         {"url": "https://example.com/bar.csv", "sha256": "ab" * 32},
...     ]
... )
>>> fs = list(src._list_files())
>>> fs[0].rel_path, fs[0].sha256
('lookups/foo.csv', None)
>>> fs[1].rel_path, fs[1].sha256 == "ab" * 32
('bar.csv', True)
>>> src.close()

URLs without a path component fall back to "index.html":

>>> with HTTPSource(urls=["https://example.com/"]) as src:
...     [r.rel_path for r in src._list_files()]
['index.html']
Source code in MEDS_extract/download/backends/http.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
class HTTPSource(Source):
    """A :class:`Source` backed by an explicit list of HTTP URLs.

    Use this for shared metadata downloads where the file list is known up-front — e.g.
    MIMIC's ``common:`` block of concept-map CSVs from ``raw.githubusercontent.com``. No
    crawling, no manifest parsing.

    Each URL entry can be either a plain string or a dict with optional ``sha256``
    and ``rel_path`` fields. ``rel_path`` defaults to the URL's basename.

    Args:
        urls: List of URL entries — plain strings or dicts. Subclasses that discover URLs
            at :meth:`_list_files` time (e.g. :class:`PhysioNetSource`) may pass ``None``.
        client: Optional pre-built :class:`httpx.Client`. When omitted, one is built via
            :meth:`_make_client` with the remaining kwargs.
        auth, headers, timeout, transport: Forwarded to :meth:`_make_client` when
            ``client`` is not provided. ``headers`` is a ``{name: value}``
            mapping applied as default headers on every request — used for API-key
            auth (``X-Dataverse-key``, bearer tokens) and content negotiation
            (``Accept:``).
        max_attempts, retry_wait: Govern the shared retry policy
            (:meth:`_retrying`) applied to both manifest GETs (:meth:`_get`) and
            streaming downloads (:meth:`_pull`) — regardless of whether ``client``
            was injected.
        include, exclude: Optional :mod:`fnmatch` globs applied to the manifest —
            see :class:`~MEDS_extract.download.source.Source`.

    Examples:
        Plain string URLs resolve to basename-based relative paths:

        >>> src = HTTPSource(urls=["https://example.com/foo.csv", "https://example.com/bar.csv"])
        >>> [r.rel_path for r in src._list_files()]
        ['foo.csv', 'bar.csv']
        >>> src.close()

        Dict entries can override ``rel_path`` and provide a checksum:

        >>> src = HTTPSource(
        ...     urls=[
        ...         {"url": "https://example.com/foo.csv", "rel_path": "lookups/foo.csv"},
        ...         {"url": "https://example.com/bar.csv", "sha256": "ab" * 32},
        ...     ]
        ... )
        >>> fs = list(src._list_files())
        >>> fs[0].rel_path, fs[0].sha256
        ('lookups/foo.csv', None)
        >>> fs[1].rel_path, fs[1].sha256 == "ab" * 32
        ('bar.csv', True)
        >>> src.close()

        URLs without a path component fall back to ``"index.html"``:

        >>> with HTTPSource(urls=["https://example.com/"]) as src:
        ...     [r.rel_path for r in src._list_files()]
        ['index.html']
    """

    # Retried via tenacity. 4xx errors are NOT retried (including 429) — those usually
    # mean the URL is wrong or auth failed, and retrying makes things worse. If
    # per-endpoint 429-with-Retry-After handling is needed later, wire it in as a
    # separate retry predicate rather than expanding this list.
    _RETRY_EXC: tuple[type[BaseException], ...] = (
        httpx.ConnectTimeout,
        httpx.ReadTimeout,
        httpx.WriteTimeout,
        httpx.PoolTimeout,
        httpx.ConnectError,
        # A mid-body TCP reset surfaces as ReadError (WriteError for uploads),
        # not RemoteProtocolError — both are as transient as the timeouts above.
        httpx.ReadError,
        httpx.WriteError,
        httpx.RemoteProtocolError,
    )

    def __init__(
        self,
        urls: list[str | dict] | None = None,
        client: httpx.Client | None = None,
        auth: tuple[str, str] | None = None,
        headers: dict[str, str] | None = None,
        timeout: tuple[float, float] = (10.0, 60.0),
        max_attempts: int = 5,
        transport: httpx.BaseTransport | None = None,
        retry_wait: wait_base | None = None,
        include: list[str] | None = None,
        exclude: list[str] | None = None,
    ):
        super().__init__(include=include, exclude=exclude)
        self._entries = [self._normalize(u) for u in (urls or [])]
        self._max_attempts = max_attempts
        self._retry_wait = retry_wait if retry_wait is not None else _RETRY_WAIT
        # Track client ownership: only close clients we built ourselves. An injected
        # ``client`` is the caller's to manage — typically tests with a shared
        # ``MockTransport``, which is reused across calls.
        self._owns_client = client is None
        self._client = (
            client
            if client is not None
            else self._make_client(auth=auth, headers=headers, timeout=timeout, transport=transport)
        )

    def _list_files(self) -> Iterable[RemoteFile]:
        yield from self._entries

    def _retrying(self) -> Retrying:
        """The shared retry policy for both request paths (``_get`` and ``_pull``).

        Built from ``self._max_attempts`` / ``self._retry_wait``, so it applies
        identically whether the httpx client was built by :meth:`_make_client` or
        injected via ``client=``. Each backoff sleep logs a WARNING naming the
        exception and wait time, so retries are distinguishable from a hang.
        """
        return Retrying(
            stop=stop_after_attempt(self._max_attempts),
            wait=self._retry_wait,
            retry=retry_if_exception(self._should_retry),
            before_sleep=before_sleep_log(logger, logging.WARNING),
            reraise=True,
        )

    def _get(self, url: str) -> httpx.Response:
        """Manifest-style GET with the source's retry policy applied.

        Raises inside the retry loop only on 5xx (so tenacity retries alongside
        the transient transport errors); 4xx responses are returned unwrapped and
        the caller decides — typically via ``raise_for_status()`` — so a bad URL
        or bad auth fails fast rather than being retried.

        Examples:
            5xx responses are retried; the third attempt succeeds. This works
            identically for an injected ``client=`` — the retry policy lives on
            the source, not the client:

            >>> import httpx as _httpx
            >>> from tenacity import wait_fixed
            >>> attempts = []
            >>> def flaky_then_ok(request):
            ...     attempts.append(None)
            ...     return _httpx.Response(503 if len(attempts) < 3 else 200, text="ok")
            >>> client = _httpx.Client(transport=_httpx.MockTransport(flaky_then_ok))
            >>> src = HTTPSource(urls=[], client=client, max_attempts=5, retry_wait=wait_fixed(0))
            >>> src._get("https://example.com/x").status_code
            200
            >>> len(attempts)  # 2 retries before the 200
            3
            >>> client.close()

            4xx is not retried — the response comes back unwrapped after one
            attempt:

            >>> attempts.clear()
            >>> def always_404(request):
            ...     attempts.append(None)
            ...     return _httpx.Response(404)
            >>> client = _httpx.Client(transport=_httpx.MockTransport(always_404))
            >>> src = HTTPSource(urls=[], client=client)
            >>> src._get("https://example.com/x").status_code
            404
            >>> len(attempts)
            1
            >>> client.close()
        """

        def _once() -> httpx.Response:
            response = self._client.get(url)
            if 500 <= response.status_code < 600:
                response.raise_for_status()
            return response

        return self._retrying()(_once)

    def _pull(self, source_path: str, target: Path) -> None:
        # Retry the whole resumable-stream attempt on transient failures. A
        # request-phase failure (connect error, 5xx before any bytes arrive) simply
        # re-issues the request; a mid-body failure leaves the enlarged ``target``
        # partial in place, so the next attempt resumes via ``Range: bytes=N-``.
        self._retrying()(self._resumable_stream, self._client, source_path, target)

    @classmethod
    def _should_retry(cls, exc: BaseException) -> bool:
        """Retry transient transport errors and 5xx responses; never 4xx.

        Shared by both request paths: ``_get`` raises only on 5xx inside its
        retry loop (4xx returns unwrapped), and ``_resumable_stream`` calls
        ``raise_for_status`` on everything — so gating ``HTTPStatusError`` on
        ``status_code >= 500`` here is what keeps 404s failing fast on both.
        """
        if isinstance(exc, cls._RETRY_EXC):
            return True
        return isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code >= 500

    def close(self) -> None:
        """Close the owned httpx client; no-op if the client was injected.

        Examples:
            When no ``client=`` is injected, the source builds and owns one, and
            ``close()`` closes it. A second ``close()`` is a no-op (httpx clients
            are re-close-safe):

            >>> src = HTTPSource(urls=["https://example.com/a.csv"])
            >>> src._owns_client, src._client.is_closed
            (True, False)
            >>> src.close()
            >>> src._client.is_closed
            True
            >>> src.close()  # idempotent

            An injected client belongs to the caller — ``close()`` leaves it open:

            >>> client = httpx.Client(transport=httpx.MockTransport(lambda r: httpx.Response(200)))
            >>> src = HTTPSource(urls=["https://example.com/a.csv"], client=client)
            >>> src._owns_client
            False
            >>> src.close()
            >>> client.is_closed
            False
            >>> client.close()  # caller cleans up

            The context-manager form closes the owned client on exit:

            >>> with HTTPSource(urls=["https://example.com/a.csv"]) as src:
            ...     inner = src._client
            ...     inner.is_closed
            False
            >>> inner.is_closed
            True
        """
        if self._owns_client:
            self._client.close()

    @classmethod
    def _make_client(
        cls,
        auth: tuple[str, str] | None = None,
        headers: dict[str, str] | None = None,
        timeout: tuple[float, float] = (10.0, 60.0),
        transport: httpx.BaseTransport | None = None,
    ) -> httpx.Client:
        """Build a plain :class:`httpx.Client` — pure client construction, no retry.

        Retry lives on the *source* (:meth:`_retrying`, applied by :meth:`_get`
        and :meth:`_pull`), not on the client — that way an injected ``client=``
        gets exactly the same retry behavior as a client built here.

        Args:
            auth: Optional ``(username, password)`` for Basic auth — e.g. PhysioNet credentials.
            headers: Optional ``{name: value}`` mapping applied as default headers on every
                request the client issues (both ``_list_files`` manifest GETs and streaming
                ``.part`` downloads). Intended for API-key auth (DataVerse's
                ``X-Dataverse-key``, generic bearer tokens) and content negotiation
                (``Accept:``). ``None`` behaves like absent.
            timeout: ``(connect_timeout, read_timeout)`` in seconds.
            transport: Optional :class:`httpx.BaseTransport` override. Defaults to the
                standard HTTP transport; pass an :class:`httpx.MockTransport` to stub out
                the wire for tests without reaching into the returned client's private
                attributes.

        Examples:
            >>> client = HTTPSource._make_client()
            >>> isinstance(client, httpx.Client)
            True
            >>> client.close()

            Basic auth is threaded through unchanged:

            >>> client = HTTPSource._make_client(auth=("user", "pass"))
            >>> client.auth
            <httpx.BasicAuth object at 0x...>
            >>> client.close()

            Custom ``headers`` reach the transport on every request — the motivating case
            is DataVerse's ``X-Dataverse-key`` API-key auth, but the same kwarg covers
            bearer tokens and ``Accept:`` content negotiation:

            >>> import httpx as _httpx
            >>> seen_headers = []
            >>> def capture(request):
            ...     seen_headers.append(dict(request.headers))
            ...     return _httpx.Response(200, text="ok")
            >>> client = HTTPSource._make_client(
            ...     headers={"X-Dataverse-key": "secret-token", "Accept": "application/json"},
            ...     transport=_httpx.MockTransport(capture),
            ... )
            >>> _ = client.get("https://example.com/x")
            >>> seen_headers[0]["x-dataverse-key"]
            'secret-token'
            >>> seen_headers[0]["accept"]
            'application/json'
            >>> client.close()
        """
        connect_timeout, read_timeout = timeout
        client_kwargs: dict = {
            "auth": httpx.BasicAuth(*auth) if auth else None,
            "headers": headers,
            "timeout": httpx.Timeout(
                connect=connect_timeout, read=read_timeout, write=read_timeout, pool=60.0
            ),
            "follow_redirects": True,
        }
        if transport is not None:
            client_kwargs["transport"] = transport
        return httpx.Client(**client_kwargs)

    @staticmethod
    def _resumable_stream(
        client: httpx.Client,
        url: str,
        target: Path,
        chunk_size: int = 1024 * 1024,
    ) -> None:
        """HTTP GET that streams bytes into ``target``, with ``Range``-resume.

        If ``target`` exists, an HTTP ``Range`` request resumes from its end;
        otherwise the download starts from byte 0. On a 416, a mismatched
        ``Content-Range``, or a server that ignores ``Range`` and returns 200,
        the resume is abandoned and the download restarts from byte 0.

        Every request sends ``Accept-Encoding: identity``. Transparent
        content-coding (httpx's default ``gzip, deflate``) would make ``target``
        hold *decoded* bytes while ``Range`` offsets and ``Content-Range``
        validation operate on the *encoded* representation — a resume against a
        compressing server would then pass the offset check yet feed the
        decompressor a mid-stream fragment. Requesting the identity coding keeps
        on-wire bytes, ``target.stat().st_size``, and the manifest's SHA-256 all
        describing the same byte stream.

        Args:
            client: A configured :class:`httpx.Client` (from :meth:`_make_client`).
            url: Absolute URL to fetch.
            target: Path to write into. May already contain partial bytes from a
                prior failed attempt — those are appended to via ``Range``.
            chunk_size: Bytes per streamed chunk.

        Raises:
            httpx.HTTPStatusError: If the server returns 4xx/5xx.
            RuntimeError: If the Range-resume restart loop fails to converge —
                defense-in-depth against a future refactor breaking the loop's
                termination invariant.

        Examples:
            The basic contract: bytes from ``url`` land in ``target``, and every
            request advertises ``Accept-Encoding: identity`` (see above for why):

            >>> def echo_handler(request):
            ...     print(f"Accept-Encoding: {request.headers.get('Accept-Encoding')}")
            ...     return httpx.Response(200, content=b"hello world")
            >>> client = httpx.Client(transport=httpx.MockTransport(echo_handler))
            >>> with tempfile.TemporaryDirectory() as d:
            ...     target = Path(d) / "x.csv.part"
            ...     HTTPSource._resumable_stream(client, "https://example.com/x.csv", target)
            ...     target.read_bytes()
            Accept-Encoding: identity
            b'hello world'
            >>> client.close()

            The Range-resume / 416 / ``Content-Range``-mismatch restart behavior
            is wire-protocol machinery exercised in ``tests/test_download.py``
            (``test_resumable_stream_*``), where multi-request handler state
            machines are more readable than doctests.
        """
        resume_from = target.stat().st_size if target.exists() else 0

        # Range-resume retry loop: if the server rejects the Range or returns a mismatched
        # 206 (or the source file changed between runs, producing 416), we restart from
        # byte 0 after clearing ``target``. Without this, a mismatched 206 silently
        # appends the wrong bytes to the existing file — undetectable except by a
        # SHA-256 mismatch on the wrapper's verify step.
        #
        # Iteration cap: by construction, a single restart zeroes ``resume_from`` and the
        # next iteration's ``if resume_from and ...`` guards short-circuit all three
        # restart branches. So the loop terminates in at most 2 iterations. The
        # ``range(_MAX_RESUME_ATTEMPTS)`` cap is defense-in-depth against a future
        # refactor breaking that invariant (e.g. someone dropping the ``resume_from = 0``
        # assignment) — better an explicit RuntimeError than a silent infinite loop.
        for _ in range(_MAX_RESUME_ATTEMPTS):
            headers = {"Accept-Encoding": "identity"}
            if resume_from:
                headers["Range"] = f"bytes={resume_from}-"
            with client.stream("GET", url, headers=headers) as r:
                # 416 "Range Not Satisfiable" — remote file shrank or changed; restart.
                if resume_from and r.status_code == 416:
                    logger.warning(
                        f"Server rejected resume for {_redact_url(url)} with 416; restarting from byte 0."
                    )
                    if target.exists():
                        target.unlink()
                    resume_from = 0
                    continue
                r.raise_for_status()
                if resume_from:
                    # Server ignored Range (200 instead of 206) → restart.
                    if r.status_code == 200:
                        # WARNING for consistency with the 416 and Content-Range
                        # siblings — all three discard the accumulated partial and
                        # re-transfer from byte 0.
                        logger.warning(
                            f"Server ignored Range for {_redact_url(url)}; restarting from byte 0."
                        )
                        if target.exists():
                            target.unlink()
                        resume_from = 0
                        continue
                    # Validate Content-Range starts at our requested offset. Without this,
                    # a server returning 206 with a shifted range silently corrupts ``target``.
                    if not HTTPSource._content_range_starts_at(r.headers.get("Content-Range"), resume_from):
                        logger.warning(
                            f"Server returned mismatched Content-Range for {_redact_url(url)} "
                            f"(got {r.headers.get('Content-Range')!r} for "
                            f"resume_from={resume_from}); restarting from byte 0."
                        )
                        if target.exists():
                            target.unlink()
                        resume_from = 0
                        continue
                mode = "ab" if resume_from else "wb"
                with target.open(mode) as f:
                    for chunk in r.iter_bytes(chunk_size=chunk_size):
                        f.write(chunk)
            return
        # Exhausted the iteration cap without a successful write+return — a bug
        # elsewhere broke the "restart zeros resume_from" invariant that makes the
        # loop terminate. Surface it loudly rather than looping forever.
        raise RuntimeError(
            f"_resumable_stream exhausted {_MAX_RESUME_ATTEMPTS} restart attempts "
            f"for {_redact_url(url)}; range-resume loop failed to converge. This indicates a bug "
            "in the restart logic — the expected invariant is that each restart "
            "resets resume_from to 0, which prevents any subsequent restart."
        )

    @staticmethod
    def _content_range_starts_at(header: str | None, expected_start: int) -> bool:
        """Parse an HTTP ``Content-Range`` header and verify it begins at ``expected_start``.

        ``Content-Range: bytes <start>-<end>/<total>`` (per RFC 7233). Only valid when the
        server sends a 206 Partial Content response. Returns ``False`` for a missing,
        malformed, or mismatched header — the caller is expected to restart the download
        on ``False``.

        Examples:
            >>> HTTPSource._content_range_starts_at("bytes 100-999/10000", 100)
            True
            >>> HTTPSource._content_range_starts_at("bytes 100-999/10000", 200)  # start mismatch
            False
            >>> HTTPSource._content_range_starts_at(None, 100)  # missing header
            False
            >>> HTTPSource._content_range_starts_at("garbage", 100)  # malformed
            False
            >>> HTTPSource._content_range_starts_at("bytes */10000", 100)  # unsatisfied-range
            False
        """
        if not header or not header.startswith("bytes "):
            return False
        range_spec = header[6:].split("/", 1)[0]
        start_str, sep, _end = range_spec.partition("-")
        return bool(sep) and start_str.isdigit() and int(start_str) == expected_start

    @staticmethod
    def _normalize(entry: str | dict) -> RemoteFile:
        """Normalize a URL entry to a validated :class:`RemoteFile`.

        Unknown dict keys are rejected rather than silently dropped — a typo like
        ``sha_256:`` would otherwise leave the download unverified while the user
        believes they pinned a checksum. Digest format/case validation happens in
        :class:`RemoteFile` itself.

        Examples:
            >>> HTTPSource._normalize("https://example.com/foo.csv")
            RemoteFile(rel_path='foo.csv', source_path='https://example.com/foo.csv', sha256=None,
                       unarchive=None, cleanup_archive=None)

            >>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha256": "ab" * 32})
            RemoteFile(rel_path='foo.csv', source_path='https://example.com/foo.csv', sha256='abab...

            Explicit ``rel_path`` wins over the URL-derived default:

            >>> HTTPSource._normalize(
            ...     {"url": "https://example.com/foo.csv", "rel_path": "lookups/foo.csv"}
            ... )
            RemoteFile(rel_path='lookups/foo.csv', source_path='https://example.com/foo.csv', sha256=None,
                       unarchive=None, cleanup_archive=None)

            Per-entry ``unarchive`` / ``cleanup_archive`` pass through to the
            :class:`RemoteFile` — the motivating case is a dataset shipped as one
            archive bundle that should be unpacked into ``dest_dir`` and discarded:

            >>> r = HTTPSource._normalize({
            ...     "url": "https://example.com/AUMCdb.zip",
            ...     "unarchive": "zip",
            ...     "cleanup_archive": True,
            ... })
            >>> r.rel_path, r.unarchive, r.cleanup_archive
            ('AUMCdb.zip', 'zip', True)

            Raises on missing ``url``, unknown keys, malformed digests, or bad type.
            The dict-shaped errors echo key names only — entry values may carry
            resolved credentials (e.g. a mis-indented ``headers:`` block) — and any
            echoed url has its userinfo masked:

            >>> HTTPSource._normalize({"sha256": "ab" * 32})
            Traceback (most recent call last):
                ...
            ValueError: HTTPSource url entry is missing 'url'; got keys ['sha256']
            >>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha_256": "ab" * 32})
            Traceback (most recent call last):
                ...
            ValueError: HTTPSource url entry has unknown keys ['sha_256'] ...
            >>> HTTPSource._normalize({"url": "https://u:pw@example.com/foo.csv", "headers": {"a": "b"}})
            Traceback (most recent call last):
                ...
            ValueError: HTTPSource url entry has unknown keys ['headers'] ... for url
                        https://***@example.com/foo.csv
            >>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha256": "abc"})
            Traceback (most recent call last):
                ...
            ValueError: sha256 must be 64 hex chars, got 'abc'
            >>> HTTPSource._normalize(42)
            Traceback (most recent call last):
                ...
            TypeError: HTTPSource url entry must be a str or dict, got int: 42
        """
        if isinstance(entry, str):
            return RemoteFile(rel_path=HTTPSource._filename_from_url(entry), source_path=entry)
        if isinstance(entry, dict):
            # Echo key names (plus a userinfo-redacted url) only, never entry values:
            # a mis-indented ``headers:`` block reaching here would otherwise put its
            # resolved token on stderr and in the persisted Hydra log.
            if "url" not in entry:
                raise ValueError(f"HTTPSource url entry is missing 'url'; got keys {sorted(entry)}")
            unknown = sorted(set(entry) - {"url", "rel_path", "sha256", "unarchive", "cleanup_archive"})
            if unknown:
                raise ValueError(
                    f"HTTPSource url entry has unknown keys {unknown} "
                    f"(supported: url, rel_path, sha256, unarchive, cleanup_archive) "
                    f"for url {_redact_url(entry['url'])}"
                )
            return RemoteFile(
                rel_path=entry.get("rel_path") or HTTPSource._filename_from_url(entry["url"]),
                source_path=entry["url"],
                sha256=entry.get("sha256"),
                unarchive=entry.get("unarchive"),
                cleanup_archive=entry.get("cleanup_archive"),  # tri-state: None defers to unarchive mode
            )
        raise TypeError(f"HTTPSource url entry must be a str or dict, got {type(entry).__name__}: {entry}")

    @staticmethod
    def _filename_from_url(url: str) -> str:
        """Derive a filesystem-friendly rel_path from ``url``.

        Examples:
            >>> HTTPSource._filename_from_url("https://example.com/foo.csv")
            'foo.csv'
            >>> HTTPSource._filename_from_url("https://example.com/path/to/bar.csv.gz")
            'bar.csv.gz'
            >>> HTTPSource._filename_from_url("https://example.com/")
            'index.html'
            >>> HTTPSource._filename_from_url("https://example.com")
            'index.html'
        """
        return Path(urlparse(url).path).name or "index.html"

_content_range_starts_at(header, expected_start) staticmethod

Parse an HTTP Content-Range header and verify it begins at expected_start.

Content-Range: bytes <start>-<end>/<total> (per RFC 7233). Only valid when the server sends a 206 Partial Content response. Returns False for a missing, malformed, or mismatched header — the caller is expected to restart the download on False.

Examples:

>>> HTTPSource._content_range_starts_at("bytes 100-999/10000", 100)
True
>>> HTTPSource._content_range_starts_at("bytes 100-999/10000", 200)  # start mismatch
False
>>> HTTPSource._content_range_starts_at(None, 100)  # missing header
False
>>> HTTPSource._content_range_starts_at("garbage", 100)  # malformed
False
>>> HTTPSource._content_range_starts_at("bytes */10000", 100)  # unsatisfied-range
False
Source code in MEDS_extract/download/backends/http.py 
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
@staticmethod
def _content_range_starts_at(header: str | None, expected_start: int) -> bool:
    """Parse an HTTP ``Content-Range`` header and verify it begins at ``expected_start``.

    ``Content-Range: bytes <start>-<end>/<total>`` (per RFC 7233). Only valid when the
    server sends a 206 Partial Content response. Returns ``False`` for a missing,
    malformed, or mismatched header — the caller is expected to restart the download
    on ``False``.

    Examples:
        >>> HTTPSource._content_range_starts_at("bytes 100-999/10000", 100)
        True
        >>> HTTPSource._content_range_starts_at("bytes 100-999/10000", 200)  # start mismatch
        False
        >>> HTTPSource._content_range_starts_at(None, 100)  # missing header
        False
        >>> HTTPSource._content_range_starts_at("garbage", 100)  # malformed
        False
        >>> HTTPSource._content_range_starts_at("bytes */10000", 100)  # unsatisfied-range
        False
    """
    if not header or not header.startswith("bytes "):
        return False
    range_spec = header[6:].split("/", 1)[0]
    start_str, sep, _end = range_spec.partition("-")
    return bool(sep) and start_str.isdigit() and int(start_str) == expected_start

_filename_from_url(url) staticmethod

Derive a filesystem-friendly rel_path from url.

Examples:

>>> HTTPSource._filename_from_url("https://example.com/foo.csv")
'foo.csv'
>>> HTTPSource._filename_from_url("https://example.com/path/to/bar.csv.gz")
'bar.csv.gz'
>>> HTTPSource._filename_from_url("https://example.com/")
'index.html'
>>> HTTPSource._filename_from_url("https://example.com")
'index.html'
Source code in MEDS_extract/download/backends/http.py 
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
@staticmethod
def _filename_from_url(url: str) -> str:
    """Derive a filesystem-friendly rel_path from ``url``.

    Examples:
        >>> HTTPSource._filename_from_url("https://example.com/foo.csv")
        'foo.csv'
        >>> HTTPSource._filename_from_url("https://example.com/path/to/bar.csv.gz")
        'bar.csv.gz'
        >>> HTTPSource._filename_from_url("https://example.com/")
        'index.html'
        >>> HTTPSource._filename_from_url("https://example.com")
        'index.html'
    """
    return Path(urlparse(url).path).name or "index.html"

_get(url)

Manifest-style GET with the source’s retry policy applied.

Raises inside the retry loop only on 5xx (so tenacity retries alongside the transient transport errors); 4xx responses are returned unwrapped and the caller decides — typically via raise_for_status() — so a bad URL or bad auth fails fast rather than being retried.

Examples:

5xx responses are retried; the third attempt succeeds. This works identically for an injected client= — the retry policy lives on the source, not the client:

>>> import httpx as _httpx
>>> from tenacity import wait_fixed
>>> attempts = []
>>> def flaky_then_ok(request):
...     attempts.append(None)
...     return _httpx.Response(503 if len(attempts) < 3 else 200, text="ok")
>>> client = _httpx.Client(transport=_httpx.MockTransport(flaky_then_ok))
>>> src = HTTPSource(urls=[], client=client, max_attempts=5, retry_wait=wait_fixed(0))
>>> src._get("https://example.com/x").status_code
200
>>> len(attempts)  # 2 retries before the 200
3
>>> client.close()

4xx is not retried — the response comes back unwrapped after one attempt:

>>> attempts.clear()
>>> def always_404(request):
...     attempts.append(None)
...     return _httpx.Response(404)
>>> client = _httpx.Client(transport=_httpx.MockTransport(always_404))
>>> src = HTTPSource(urls=[], client=client)
>>> src._get("https://example.com/x").status_code
404
>>> len(attempts)
1
>>> client.close()
Source code in MEDS_extract/download/backends/http.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def _get(self, url: str) -> httpx.Response:
    """Manifest-style GET with the source's retry policy applied.

    Raises inside the retry loop only on 5xx (so tenacity retries alongside
    the transient transport errors); 4xx responses are returned unwrapped and
    the caller decides — typically via ``raise_for_status()`` — so a bad URL
    or bad auth fails fast rather than being retried.

    Examples:
        5xx responses are retried; the third attempt succeeds. This works
        identically for an injected ``client=`` — the retry policy lives on
        the source, not the client:

        >>> import httpx as _httpx
        >>> from tenacity import wait_fixed
        >>> attempts = []
        >>> def flaky_then_ok(request):
        ...     attempts.append(None)
        ...     return _httpx.Response(503 if len(attempts) < 3 else 200, text="ok")
        >>> client = _httpx.Client(transport=_httpx.MockTransport(flaky_then_ok))
        >>> src = HTTPSource(urls=[], client=client, max_attempts=5, retry_wait=wait_fixed(0))
        >>> src._get("https://example.com/x").status_code
        200
        >>> len(attempts)  # 2 retries before the 200
        3
        >>> client.close()

        4xx is not retried — the response comes back unwrapped after one
        attempt:

        >>> attempts.clear()
        >>> def always_404(request):
        ...     attempts.append(None)
        ...     return _httpx.Response(404)
        >>> client = _httpx.Client(transport=_httpx.MockTransport(always_404))
        >>> src = HTTPSource(urls=[], client=client)
        >>> src._get("https://example.com/x").status_code
        404
        >>> len(attempts)
        1
        >>> client.close()
    """

    def _once() -> httpx.Response:
        response = self._client.get(url)
        if 500 <= response.status_code < 600:
            response.raise_for_status()
        return response

    return self._retrying()(_once)

_make_client(auth=None, headers=None, timeout=(10.0, 60.0), transport=None) classmethod

Build a plain :class:httpx.Client — pure client construction, no retry.

Retry lives on the source (:meth:_retrying, applied by :meth:_get and :meth:_pull), not on the client — that way an injected client= gets exactly the same retry behavior as a client built here.

Parameters:

Name Type Description Default
auth tuple[str, str] | None

Optional (username, password) for Basic auth — e.g. PhysioNet credentials.

 None
headers dict[str, str] | None

Optional {name: value} mapping applied as default headers on every request the client issues (both _list_files manifest GETs and streaming .part downloads). Intended for API-key auth (DataVerse’s X-Dataverse-key, generic bearer tokens) and content negotiation (Accept:). None behaves like absent.

 None
timeout tuple[float, float]

(connect_timeout, read_timeout) in seconds.

 (10.0, 60.0)
transport BaseTransport | None

Optional :class:httpx.BaseTransport override. Defaults to the standard HTTP transport; pass an :class:httpx.MockTransport to stub out the wire for tests without reaching into the returned client’s private attributes.

 None

Examples:

>>> client = HTTPSource._make_client()
>>> isinstance(client, httpx.Client)
True
>>> client.close()

Basic auth is threaded through unchanged:

>>> client = HTTPSource._make_client(auth=("user", "pass"))
>>> client.auth
<httpx.BasicAuth object at 0x...>
>>> client.close()

Custom headers reach the transport on every request — the motivating case is DataVerse’s X-Dataverse-key API-key auth, but the same kwarg covers bearer tokens and Accept: content negotiation:

>>> import httpx as _httpx
>>> seen_headers = []
>>> def capture(request):
...     seen_headers.append(dict(request.headers))
...     return _httpx.Response(200, text="ok")
>>> client = HTTPSource._make_client(
...     headers={"X-Dataverse-key": "secret-token", "Accept": "application/json"},
...     transport=_httpx.MockTransport(capture),
... )
>>> _ = client.get("https://example.com/x")
>>> seen_headers[0]["x-dataverse-key"]
'secret-token'
>>> seen_headers[0]["accept"]
'application/json'
>>> client.close()
Source code in MEDS_extract/download/backends/http.py 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
@classmethod
def _make_client(
    cls,
    auth: tuple[str, str] | None = None,
    headers: dict[str, str] | None = None,
    timeout: tuple[float, float] = (10.0, 60.0),
    transport: httpx.BaseTransport | None = None,
) -> httpx.Client:
    """Build a plain :class:`httpx.Client` — pure client construction, no retry.

    Retry lives on the *source* (:meth:`_retrying`, applied by :meth:`_get`
    and :meth:`_pull`), not on the client — that way an injected ``client=``
    gets exactly the same retry behavior as a client built here.

    Args:
        auth: Optional ``(username, password)`` for Basic auth — e.g. PhysioNet credentials.
        headers: Optional ``{name: value}`` mapping applied as default headers on every
            request the client issues (both ``_list_files`` manifest GETs and streaming
            ``.part`` downloads). Intended for API-key auth (DataVerse's
            ``X-Dataverse-key``, generic bearer tokens) and content negotiation
            (``Accept:``). ``None`` behaves like absent.
        timeout: ``(connect_timeout, read_timeout)`` in seconds.
        transport: Optional :class:`httpx.BaseTransport` override. Defaults to the
            standard HTTP transport; pass an :class:`httpx.MockTransport` to stub out
            the wire for tests without reaching into the returned client's private
            attributes.

    Examples:
        >>> client = HTTPSource._make_client()
        >>> isinstance(client, httpx.Client)
        True
        >>> client.close()

        Basic auth is threaded through unchanged:

        >>> client = HTTPSource._make_client(auth=("user", "pass"))
        >>> client.auth
        <httpx.BasicAuth object at 0x...>
        >>> client.close()

        Custom ``headers`` reach the transport on every request — the motivating case
        is DataVerse's ``X-Dataverse-key`` API-key auth, but the same kwarg covers
        bearer tokens and ``Accept:`` content negotiation:

        >>> import httpx as _httpx
        >>> seen_headers = []
        >>> def capture(request):
        ...     seen_headers.append(dict(request.headers))
        ...     return _httpx.Response(200, text="ok")
        >>> client = HTTPSource._make_client(
        ...     headers={"X-Dataverse-key": "secret-token", "Accept": "application/json"},
        ...     transport=_httpx.MockTransport(capture),
        ... )
        >>> _ = client.get("https://example.com/x")
        >>> seen_headers[0]["x-dataverse-key"]
        'secret-token'
        >>> seen_headers[0]["accept"]
        'application/json'
        >>> client.close()
    """
    connect_timeout, read_timeout = timeout
    client_kwargs: dict = {
        "auth": httpx.BasicAuth(*auth) if auth else None,
        "headers": headers,
        "timeout": httpx.Timeout(
            connect=connect_timeout, read=read_timeout, write=read_timeout, pool=60.0
        ),
        "follow_redirects": True,
    }
    if transport is not None:
        client_kwargs["transport"] = transport
    return httpx.Client(**client_kwargs)

_normalize(entry) staticmethod

Normalize a URL entry to a validated :class:RemoteFile.

Unknown dict keys are rejected rather than silently dropped — a typo like sha_256: would otherwise leave the download unverified while the user believes they pinned a checksum. Digest format/case validation happens in :class:RemoteFile itself.

Examples:

>>> HTTPSource._normalize("https://example.com/foo.csv")
RemoteFile(rel_path='foo.csv', source_path='https://example.com/foo.csv', sha256=None,
           unarchive=None, cleanup_archive=None)

>>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha256": "ab" * 32})
RemoteFile(rel_path='foo.csv', source_path='https://example.com/foo.csv', sha256='abab...

Explicit rel_path wins over the URL-derived default:

>>> HTTPSource._normalize(
...     {"url": "https://example.com/foo.csv", "rel_path": "lookups/foo.csv"}
... )
RemoteFile(rel_path='lookups/foo.csv', source_path='https://example.com/foo.csv', sha256=None,
           unarchive=None, cleanup_archive=None)

Per-entry unarchive / cleanup_archive pass through to the :class:RemoteFile — the motivating case is a dataset shipped as one archive bundle that should be unpacked into dest_dir and discarded:

>>> r = HTTPSource._normalize({
...     "url": "https://example.com/AUMCdb.zip",
...     "unarchive": "zip",
...     "cleanup_archive": True,
... })
>>> r.rel_path, r.unarchive, r.cleanup_archive
('AUMCdb.zip', 'zip', True)

Raises on missing url, unknown keys, malformed digests, or bad type. The dict-shaped errors echo key names only — entry values may carry resolved credentials (e.g. a mis-indented headers: block) — and any echoed url has its userinfo masked:

>>> HTTPSource._normalize({"sha256": "ab" * 32})
Traceback (most recent call last):
    ...
ValueError: HTTPSource url entry is missing 'url'; got keys ['sha256']
>>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha_256": "ab" * 32})
Traceback (most recent call last):
    ...
ValueError: HTTPSource url entry has unknown keys ['sha_256'] ...
>>> HTTPSource._normalize({"url": "https://u:pw@example.com/foo.csv", "headers": {"a": "b"}})
Traceback (most recent call last):
    ...
ValueError: HTTPSource url entry has unknown keys ['headers'] ... for url
            https://***@example.com/foo.csv
>>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha256": "abc"})
Traceback (most recent call last):
    ...
ValueError: sha256 must be 64 hex chars, got 'abc'
>>> HTTPSource._normalize(42)
Traceback (most recent call last):
    ...
TypeError: HTTPSource url entry must be a str or dict, got int: 42
Source code in MEDS_extract/download/backends/http.py 
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
@staticmethod
def _normalize(entry: str | dict) -> RemoteFile:
    """Normalize a URL entry to a validated :class:`RemoteFile`.

    Unknown dict keys are rejected rather than silently dropped — a typo like
    ``sha_256:`` would otherwise leave the download unverified while the user
    believes they pinned a checksum. Digest format/case validation happens in
    :class:`RemoteFile` itself.

    Examples:
        >>> HTTPSource._normalize("https://example.com/foo.csv")
        RemoteFile(rel_path='foo.csv', source_path='https://example.com/foo.csv', sha256=None,
                   unarchive=None, cleanup_archive=None)

        >>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha256": "ab" * 32})
        RemoteFile(rel_path='foo.csv', source_path='https://example.com/foo.csv', sha256='abab...

        Explicit ``rel_path`` wins over the URL-derived default:

        >>> HTTPSource._normalize(
        ...     {"url": "https://example.com/foo.csv", "rel_path": "lookups/foo.csv"}
        ... )
        RemoteFile(rel_path='lookups/foo.csv', source_path='https://example.com/foo.csv', sha256=None,
                   unarchive=None, cleanup_archive=None)

        Per-entry ``unarchive`` / ``cleanup_archive`` pass through to the
        :class:`RemoteFile` — the motivating case is a dataset shipped as one
        archive bundle that should be unpacked into ``dest_dir`` and discarded:

        >>> r = HTTPSource._normalize({
        ...     "url": "https://example.com/AUMCdb.zip",
        ...     "unarchive": "zip",
        ...     "cleanup_archive": True,
        ... })
        >>> r.rel_path, r.unarchive, r.cleanup_archive
        ('AUMCdb.zip', 'zip', True)

        Raises on missing ``url``, unknown keys, malformed digests, or bad type.
        The dict-shaped errors echo key names only — entry values may carry
        resolved credentials (e.g. a mis-indented ``headers:`` block) — and any
        echoed url has its userinfo masked:

        >>> HTTPSource._normalize({"sha256": "ab" * 32})
        Traceback (most recent call last):
            ...
        ValueError: HTTPSource url entry is missing 'url'; got keys ['sha256']
        >>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha_256": "ab" * 32})
        Traceback (most recent call last):
            ...
        ValueError: HTTPSource url entry has unknown keys ['sha_256'] ...
        >>> HTTPSource._normalize({"url": "https://u:pw@example.com/foo.csv", "headers": {"a": "b"}})
        Traceback (most recent call last):
            ...
        ValueError: HTTPSource url entry has unknown keys ['headers'] ... for url
                    https://***@example.com/foo.csv
        >>> HTTPSource._normalize({"url": "https://example.com/foo.csv", "sha256": "abc"})
        Traceback (most recent call last):
            ...
        ValueError: sha256 must be 64 hex chars, got 'abc'
        >>> HTTPSource._normalize(42)
        Traceback (most recent call last):
            ...
        TypeError: HTTPSource url entry must be a str or dict, got int: 42
    """
    if isinstance(entry, str):
        return RemoteFile(rel_path=HTTPSource._filename_from_url(entry), source_path=entry)
    if isinstance(entry, dict):
        # Echo key names (plus a userinfo-redacted url) only, never entry values:
        # a mis-indented ``headers:`` block reaching here would otherwise put its
        # resolved token on stderr and in the persisted Hydra log.
        if "url" not in entry:
            raise ValueError(f"HTTPSource url entry is missing 'url'; got keys {sorted(entry)}")
        unknown = sorted(set(entry) - {"url", "rel_path", "sha256", "unarchive", "cleanup_archive"})
        if unknown:
            raise ValueError(
                f"HTTPSource url entry has unknown keys {unknown} "
                f"(supported: url, rel_path, sha256, unarchive, cleanup_archive) "
                f"for url {_redact_url(entry['url'])}"
            )
        return RemoteFile(
            rel_path=entry.get("rel_path") or HTTPSource._filename_from_url(entry["url"]),
            source_path=entry["url"],
            sha256=entry.get("sha256"),
            unarchive=entry.get("unarchive"),
            cleanup_archive=entry.get("cleanup_archive"),  # tri-state: None defers to unarchive mode
        )
    raise TypeError(f"HTTPSource url entry must be a str or dict, got {type(entry).__name__}: {entry}")

_resumable_stream(client, url, target, chunk_size=1024 * 1024) staticmethod

HTTP GET that streams bytes into target, with Range-resume.

If target exists, an HTTP Range request resumes from its end; otherwise the download starts from byte 0. On a 416, a mismatched Content-Range, or a server that ignores Range and returns 200, the resume is abandoned and the download restarts from byte 0.

Every request sends Accept-Encoding: identity. Transparent content-coding (httpx’s default gzip, deflate) would make target hold decoded bytes while Range offsets and Content-Range validation operate on the encoded representation — a resume against a compressing server would then pass the offset check yet feed the decompressor a mid-stream fragment. Requesting the identity coding keeps on-wire bytes, target.stat().st_size, and the manifest’s SHA-256 all describing the same byte stream.

Parameters:

Name Type Description Default
client Client

A configured :class:httpx.Client (from :meth:_make_client).

 required
url str

Absolute URL to fetch.

 required
target Path

Path to write into. May already contain partial bytes from a prior failed attempt — those are appended to via Range.

 required
chunk_size int

Bytes per streamed chunk.

 1024 * 1024

Raises:

Type Description
HTTPStatusError

If the server returns 4xx/5xx.
RuntimeError

If the Range-resume restart loop fails to converge — defense-in-depth against a future refactor breaking the loop’s termination invariant.

Examples:

The basic contract: bytes from url land in target, and every request advertises Accept-Encoding: identity (see above for why):

>>> def echo_handler(request):
...     print(f"Accept-Encoding: {request.headers.get('Accept-Encoding')}")
...     return httpx.Response(200, content=b"hello world")
>>> client = httpx.Client(transport=httpx.MockTransport(echo_handler))
>>> with tempfile.TemporaryDirectory() as d:
...     target = Path(d) / "x.csv.part"
...     HTTPSource._resumable_stream(client, "https://example.com/x.csv", target)
...     target.read_bytes()
Accept-Encoding: identity
b'hello world'
>>> client.close()

The Range-resume / 416 / Content-Range-mismatch restart behavior is wire-protocol machinery exercised in tests/test_download.py (test_resumable_stream_*), where multi-request handler state machines are more readable than doctests.

Source code in MEDS_extract/download/backends/http.py 
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
@staticmethod
def _resumable_stream(
    client: httpx.Client,
    url: str,
    target: Path,
    chunk_size: int = 1024 * 1024,
) -> None:
    """HTTP GET that streams bytes into ``target``, with ``Range``-resume.

    If ``target`` exists, an HTTP ``Range`` request resumes from its end;
    otherwise the download starts from byte 0. On a 416, a mismatched
    ``Content-Range``, or a server that ignores ``Range`` and returns 200,
    the resume is abandoned and the download restarts from byte 0.

    Every request sends ``Accept-Encoding: identity``. Transparent
    content-coding (httpx's default ``gzip, deflate``) would make ``target``
    hold *decoded* bytes while ``Range`` offsets and ``Content-Range``
    validation operate on the *encoded* representation — a resume against a
    compressing server would then pass the offset check yet feed the
    decompressor a mid-stream fragment. Requesting the identity coding keeps
    on-wire bytes, ``target.stat().st_size``, and the manifest's SHA-256 all
    describing the same byte stream.

    Args:
        client: A configured :class:`httpx.Client` (from :meth:`_make_client`).
        url: Absolute URL to fetch.
        target: Path to write into. May already contain partial bytes from a
            prior failed attempt — those are appended to via ``Range``.
        chunk_size: Bytes per streamed chunk.

    Raises:
        httpx.HTTPStatusError: If the server returns 4xx/5xx.
        RuntimeError: If the Range-resume restart loop fails to converge —
            defense-in-depth against a future refactor breaking the loop's
            termination invariant.

    Examples:
        The basic contract: bytes from ``url`` land in ``target``, and every
        request advertises ``Accept-Encoding: identity`` (see above for why):

        >>> def echo_handler(request):
        ...     print(f"Accept-Encoding: {request.headers.get('Accept-Encoding')}")
        ...     return httpx.Response(200, content=b"hello world")
        >>> client = httpx.Client(transport=httpx.MockTransport(echo_handler))
        >>> with tempfile.TemporaryDirectory() as d:
        ...     target = Path(d) / "x.csv.part"
        ...     HTTPSource._resumable_stream(client, "https://example.com/x.csv", target)
        ...     target.read_bytes()
        Accept-Encoding: identity
        b'hello world'
        >>> client.close()

        The Range-resume / 416 / ``Content-Range``-mismatch restart behavior
        is wire-protocol machinery exercised in ``tests/test_download.py``
        (``test_resumable_stream_*``), where multi-request handler state
        machines are more readable than doctests.
    """
    resume_from = target.stat().st_size if target.exists() else 0

    # Range-resume retry loop: if the server rejects the Range or returns a mismatched
    # 206 (or the source file changed between runs, producing 416), we restart from
    # byte 0 after clearing ``target``. Without this, a mismatched 206 silently
    # appends the wrong bytes to the existing file — undetectable except by a
    # SHA-256 mismatch on the wrapper's verify step.
    #
    # Iteration cap: by construction, a single restart zeroes ``resume_from`` and the
    # next iteration's ``if resume_from and ...`` guards short-circuit all three
    # restart branches. So the loop terminates in at most 2 iterations. The
    # ``range(_MAX_RESUME_ATTEMPTS)`` cap is defense-in-depth against a future
    # refactor breaking that invariant (e.g. someone dropping the ``resume_from = 0``
    # assignment) — better an explicit RuntimeError than a silent infinite loop.
    for _ in range(_MAX_RESUME_ATTEMPTS):
        headers = {"Accept-Encoding": "identity"}
        if resume_from:
            headers["Range"] = f"bytes={resume_from}-"
        with client.stream("GET", url, headers=headers) as r:
            # 416 "Range Not Satisfiable" — remote file shrank or changed; restart.
            if resume_from and r.status_code == 416:
                logger.warning(
                    f"Server rejected resume for {_redact_url(url)} with 416; restarting from byte 0."
                )
                if target.exists():
                    target.unlink()
                resume_from = 0
                continue
            r.raise_for_status()
            if resume_from:
                # Server ignored Range (200 instead of 206) → restart.
                if r.status_code == 200:
                    # WARNING for consistency with the 416 and Content-Range
                    # siblings — all three discard the accumulated partial and
                    # re-transfer from byte 0.
                    logger.warning(
                        f"Server ignored Range for {_redact_url(url)}; restarting from byte 0."
                    )
                    if target.exists():
                        target.unlink()
                    resume_from = 0
                    continue
                # Validate Content-Range starts at our requested offset. Without this,
                # a server returning 206 with a shifted range silently corrupts ``target``.
                if not HTTPSource._content_range_starts_at(r.headers.get("Content-Range"), resume_from):
                    logger.warning(
                        f"Server returned mismatched Content-Range for {_redact_url(url)} "
                        f"(got {r.headers.get('Content-Range')!r} for "
                        f"resume_from={resume_from}); restarting from byte 0."
                    )
                    if target.exists():
                        target.unlink()
                    resume_from = 0
                    continue
            mode = "ab" if resume_from else "wb"
            with target.open(mode) as f:
                for chunk in r.iter_bytes(chunk_size=chunk_size):
                    f.write(chunk)
        return
    # Exhausted the iteration cap without a successful write+return — a bug
    # elsewhere broke the "restart zeros resume_from" invariant that makes the
    # loop terminate. Surface it loudly rather than looping forever.
    raise RuntimeError(
        f"_resumable_stream exhausted {_MAX_RESUME_ATTEMPTS} restart attempts "
        f"for {_redact_url(url)}; range-resume loop failed to converge. This indicates a bug "
        "in the restart logic — the expected invariant is that each restart "
        "resets resume_from to 0, which prevents any subsequent restart."
    )

_retrying()

The shared retry policy for both request paths (_get and _pull).

Built from self._max_attempts / self._retry_wait, so it applies identically whether the httpx client was built by :meth:_make_client or injected via client=. Each backoff sleep logs a WARNING naming the exception and wait time, so retries are distinguishable from a hang.

Source code in MEDS_extract/download/backends/http.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
def _retrying(self) -> Retrying:
    """The shared retry policy for both request paths (``_get`` and ``_pull``).

    Built from ``self._max_attempts`` / ``self._retry_wait``, so it applies
    identically whether the httpx client was built by :meth:`_make_client` or
    injected via ``client=``. Each backoff sleep logs a WARNING naming the
    exception and wait time, so retries are distinguishable from a hang.
    """
    return Retrying(
        stop=stop_after_attempt(self._max_attempts),
        wait=self._retry_wait,
        retry=retry_if_exception(self._should_retry),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )

_should_retry(exc) classmethod

Retry transient transport errors and 5xx responses; never 4xx.

Shared by both request paths: _get raises only on 5xx inside its retry loop (4xx returns unwrapped), and _resumable_stream calls raise_for_status on everything — so gating HTTPStatusError on status_code >= 500 here is what keeps 404s failing fast on both.

Source code in MEDS_extract/download/backends/http.py 
280
281
282
283
284
285
286
287
288
289
290
291
@classmethod
def _should_retry(cls, exc: BaseException) -> bool:
    """Retry transient transport errors and 5xx responses; never 4xx.

    Shared by both request paths: ``_get`` raises only on 5xx inside its
    retry loop (4xx returns unwrapped), and ``_resumable_stream`` calls
    ``raise_for_status`` on everything — so gating ``HTTPStatusError`` on
    ``status_code >= 500`` here is what keeps 404s failing fast on both.
    """
    if isinstance(exc, cls._RETRY_EXC):
        return True
    return isinstance(exc, httpx.HTTPStatusError) and exc.response.status_code >= 500

close()

Close the owned httpx client; no-op if the client was injected.

Examples:

When no client= is injected, the source builds and owns one, and close() closes it. A second close() is a no-op (httpx clients are re-close-safe):

>>> src = HTTPSource(urls=["https://example.com/a.csv"])
>>> src._owns_client, src._client.is_closed
(True, False)
>>> src.close()
>>> src._client.is_closed
True
>>> src.close()  # idempotent

An injected client belongs to the caller — close() leaves it open:

>>> client = httpx.Client(transport=httpx.MockTransport(lambda r: httpx.Response(200)))
>>> src = HTTPSource(urls=["https://example.com/a.csv"], client=client)
>>> src._owns_client
False
>>> src.close()
>>> client.is_closed
False
>>> client.close()  # caller cleans up

The context-manager form closes the owned client on exit:

>>> with HTTPSource(urls=["https://example.com/a.csv"]) as src:
...     inner = src._client
...     inner.is_closed
False
>>> inner.is_closed
True
Source code in MEDS_extract/download/backends/http.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def close(self) -> None:
    """Close the owned httpx client; no-op if the client was injected.

    Examples:
        When no ``client=`` is injected, the source builds and owns one, and
        ``close()`` closes it. A second ``close()`` is a no-op (httpx clients
        are re-close-safe):

        >>> src = HTTPSource(urls=["https://example.com/a.csv"])
        >>> src._owns_client, src._client.is_closed
        (True, False)
        >>> src.close()
        >>> src._client.is_closed
        True
        >>> src.close()  # idempotent

        An injected client belongs to the caller — ``close()`` leaves it open:

        >>> client = httpx.Client(transport=httpx.MockTransport(lambda r: httpx.Response(200)))
        >>> src = HTTPSource(urls=["https://example.com/a.csv"], client=client)
        >>> src._owns_client
        False
        >>> src.close()
        >>> client.is_closed
        False
        >>> client.close()  # caller cleans up

        The context-manager form closes the owned client on exit:

        >>> with HTTPSource(urls=["https://example.com/a.csv"]) as src:
        ...     inner = src._client
        ...     inner.is_closed
        False
        >>> inner.is_closed
        True
    """
    if self._owns_client:
        self._client.close()

PhysioNetSource

Bases: HTTPSource

A :class:Source for any PhysioNet dataset release.

Inherits all HTTP machinery (client, retry, Range-resume download, checksum verify) from :class:HTTPSource — it overrides :meth:_list_files (plus its constructor, which takes a release URL and credentials instead of an explicit URL list). Uses the SHA256SUMS.txt manifest that every PhysioNet release publishes as the authoritative file list: each line is <sha256> <rel_path>, and each entry’s URL is just {base_url}/{rel_path}.

Credential plumbing for restricted datasets (MIMIC-IV, eICU, etc.) is HTTP Basic auth via the username / password kwargs; open datasets (MIMIC-IV demo) need neither. Basic auth alone is not sufficient, though: physionet.org serves credentialed /files/ paths only to clients whose User-Agent starts with Wget/<version> (their documented bulk-download tool). The gate is a prefix match — anything appended after the Wget/<version> token is preserved — and it rejects with a 403 before credentials are considered, with no WWW-Authenticate challenge, so a wrong UA is otherwise indistinguishable from wrong credentials. To pass the gate while staying honestly identified, clients built here default to Wget/<version> MEDS-Extract/<version> (see _default_user_agent); a user-supplied headers={"User-Agent": ...} wins completely.

Parameters:

Name Type Description Default
base_url str

The PhysioNet release URL, with or without trailing slash — e.g. "https://physionet.org/files/mimiciv/3.1/".

 required
username str | None

PhysioNet username (Basic auth). Omit for open-access datasets.

 None
password str | None

PhysioNet password. Omit for open-access datasets.

 None
client Client | None

Optional injected :class:httpx.Client (used by tests). When omitted, one is built via :meth:HTTPSource._make_client with the supplied auth.

 None
headers, timeout, max_attempts, transport, retry_wait

Forwarded to :meth:HTTPSource._make_client when client is not provided. Unless headers supplies its own User-Agent, the default described above is injected — physionet’s /files/ gate makes the UA load-bearing here, unlike plain :class:HTTPSource (which keeps httpx’s stock UA).

 required
include, exclude

Optional :mod:fnmatch globs applied to the manifest — e.g. include=["hosp/*.csv.gz"] stages only the hospital tables from a release that also bundles data the ETL never reads. See :class:~MEDS_extract.download.source.Source.

 required
unarchive str | None

Blanket post-fetch unpack mode applied to every :class:~MEDS_extract.download.source.RemoteFile this source lists. Typically "auto" — members whose rel_path ends in .zip / .tar.gz / .tgz / .tar get unpacked after fetch; everything else (.csv.gz, .txt, …) is a no-op. None (default) preserves the “write archive as-is” behavior.

 None
cleanup_archive bool | None

Tri-state controlling per-member archive cleanup after a successful extraction. None (default) defers to the unarchive mode — see :class:~MEDS_extract.download.source.RemoteFile. Set True / False to force the choice for every listed member.

 None

Examples:

Public releases (e.g. MIMIC-IV demo) need no auth — construction is eager but does no network I/O until the manifest is first accessed (:attr:Source.files, e.g. via :meth:Source.download_all or :func:~MEDS_extract.download.source.validate_unique_destinations):

>>> src = PhysioNetSource(base_url="https://physionet.org/files/mimic-iv-demo/2.2")
>>> src._base_url
'https://physionet.org/files/mimic-iv-demo/2.2/'
>>> src.close()

The base URL is normalized to end in exactly one trailing slash so URL concatenation is clean — an already-slashed URL passes through unchanged:

>>> with PhysioNetSource(base_url="https://physionet.org/files/mimic-iv-demo/2.2/") as src:
...     src._base_url
'https://physionet.org/files/mimic-iv-demo/2.2/'

Credentialed releases (MIMIC-IV, eICU, etc.) take username / password:

>>> src = PhysioNetSource(
...     base_url="https://physionet.org/files/mimiciv/3.1",
...     username="demo_user", password="demo_pw",
... )
>>> src.close()

Half-credentials are rejected eagerly (better to fail at construction than on first Basic-auth request):

>>> PhysioNetSource(base_url="https://physionet.org/x/1.0", username="u")
Traceback (most recent call last):
    ...
ValueError: PhysioNetSource: username and password must be supplied together ...

The reversed half — a password without a username — is rejected the same way:

>>> PhysioNetSource(base_url="https://physionet.org/x/1.0", password="p")
Traceback (most recent call last):
    ...
ValueError: PhysioNetSource: username and password must be supplied together ...

unarchive / cleanup_archive propagate to every :class:~MEDS_extract.download.source.RemoteFile listed. "auto" is the expected value for releases that ship archive members alongside non-archive ones — the unpack only fires for the actual archives:

>>> with PhysioNetSource(
...     base_url="https://physionet.org/files/example/1.0",
...     unarchive="auto",
... ) as src:
...     src._unarchive, src._cleanup_archive
('auto', None)
Source code in MEDS_extract/download/backends/physionet.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
class PhysioNetSource(HTTPSource):
    """A :class:`Source` for any PhysioNet dataset release.

    Inherits all HTTP machinery (client, retry, Range-resume download, checksum verify)
    from :class:`HTTPSource` — it overrides :meth:`_list_files` (plus its constructor,
    which takes a release URL and credentials instead of an explicit URL list). Uses
    the ``SHA256SUMS.txt`` manifest that every PhysioNet release publishes as the
    authoritative file list: each line is ``<sha256>  <rel_path>``, and each entry's URL
    is just ``{base_url}/{rel_path}``.

    Credential plumbing for restricted datasets (MIMIC-IV, eICU, etc.) is HTTP Basic auth
    via the ``username`` / ``password`` kwargs; open datasets (MIMIC-IV demo) need
    neither. Basic auth alone is **not** sufficient, though: physionet.org serves
    credentialed ``/files/`` paths only to clients whose ``User-Agent`` starts with
    ``Wget/<version>`` (their documented bulk-download tool). The gate is a prefix
    match — anything appended after the ``Wget/<version>`` token is preserved — and it
    rejects with a 403 *before* credentials are considered, with no ``WWW-Authenticate``
    challenge, so a wrong UA is otherwise indistinguishable from wrong credentials. To
    pass the gate while staying honestly identified, clients built here default to
    ``Wget/<version> MEDS-Extract/<version>`` (see ``_default_user_agent``); a
    user-supplied ``headers={"User-Agent": ...}`` wins completely.

    Args:
        base_url: The PhysioNet release URL, with or without trailing slash — e.g.
            ``"https://physionet.org/files/mimiciv/3.1/"``.
        username: PhysioNet username (Basic auth). Omit for open-access datasets.
        password: PhysioNet password. Omit for open-access datasets.
        client: Optional injected :class:`httpx.Client` (used by tests). When omitted,
            one is built via :meth:`HTTPSource._make_client` with the supplied auth.
        headers, timeout, max_attempts, transport, retry_wait: Forwarded to
            :meth:`HTTPSource._make_client` when ``client`` is not provided.
            Unless ``headers`` supplies its own ``User-Agent``, the default described
            above is injected — physionet's ``/files/`` gate makes the UA
            load-bearing here, unlike plain :class:`HTTPSource` (which keeps httpx's
            stock UA).
        include, exclude: Optional :mod:`fnmatch` globs applied to the manifest —
            e.g. ``include=["hosp/*.csv.gz"]`` stages only the hospital tables from
            a release that also bundles data the ETL never reads. See
            :class:`~MEDS_extract.download.source.Source`.
        unarchive: Blanket post-fetch unpack mode applied to every
            :class:`~MEDS_extract.download.source.RemoteFile` this source lists.
            Typically ``"auto"`` — members whose ``rel_path`` ends in ``.zip`` /
            ``.tar.gz`` / ``.tgz`` / ``.tar`` get unpacked after fetch; everything
            else (``.csv.gz``, ``.txt``, ...) is a no-op. ``None`` (default)
            preserves the "write archive as-is" behavior.
        cleanup_archive: Tri-state controlling per-member archive cleanup after a
            successful extraction. ``None`` (default) defers to the ``unarchive``
            mode — see :class:`~MEDS_extract.download.source.RemoteFile`. Set
            ``True`` / ``False`` to force the choice for every listed member.

    Examples:
        Public releases (e.g. MIMIC-IV demo) need no auth — construction is eager but
        does no network I/O until the manifest is first accessed (:attr:`Source.files`,
        e.g. via :meth:`Source.download_all` or
        :func:`~MEDS_extract.download.source.validate_unique_destinations`):

        >>> src = PhysioNetSource(base_url="https://physionet.org/files/mimic-iv-demo/2.2")
        >>> src._base_url
        'https://physionet.org/files/mimic-iv-demo/2.2/'
        >>> src.close()

        The base URL is normalized to end in exactly one trailing slash so URL
        concatenation is clean — an already-slashed URL passes through unchanged:

        >>> with PhysioNetSource(base_url="https://physionet.org/files/mimic-iv-demo/2.2/") as src:
        ...     src._base_url
        'https://physionet.org/files/mimic-iv-demo/2.2/'

        Credentialed releases (MIMIC-IV, eICU, etc.) take ``username`` / ``password``:

        >>> src = PhysioNetSource(
        ...     base_url="https://physionet.org/files/mimiciv/3.1",
        ...     username="demo_user", password="demo_pw",
        ... )
        >>> src.close()

        Half-credentials are rejected eagerly (better to fail at construction than on
        first Basic-auth request):

        >>> PhysioNetSource(base_url="https://physionet.org/x/1.0", username="u")
        Traceback (most recent call last):
            ...
        ValueError: PhysioNetSource: username and password must be supplied together ...

        The reversed half — a password without a username — is rejected the same way:

        >>> PhysioNetSource(base_url="https://physionet.org/x/1.0", password="p")
        Traceback (most recent call last):
            ...
        ValueError: PhysioNetSource: username and password must be supplied together ...

        ``unarchive`` / ``cleanup_archive`` propagate to every
        :class:`~MEDS_extract.download.source.RemoteFile` listed. ``"auto"`` is the
        expected value for releases that ship archive members alongside non-archive
        ones — the unpack only fires for the actual archives:

        >>> with PhysioNetSource(
        ...     base_url="https://physionet.org/files/example/1.0",
        ...     unarchive="auto",
        ... ) as src:
        ...     src._unarchive, src._cleanup_archive
        ('auto', None)
    """

    def __init__(
        self,
        base_url: str,
        username: str | None = None,
        password: str | None = None,
        client: httpx.Client | None = None,
        headers: dict[str, str] | None = None,
        timeout: tuple[float, float] = (10.0, 60.0),
        max_attempts: int = 5,
        transport: httpx.BaseTransport | None = None,
        retry_wait: wait_base | None = None,
        include: list[str] | None = None,
        exclude: list[str] | None = None,
        unarchive: str | None = None,
        cleanup_archive: bool | None = None,
    ):
        if (username is None) != (password is None):
            raise ValueError(
                f"{type(self).__name__}: username and password must be supplied together "
                f"(got username={username!r}, password={'<set>' if password else None!r}). "
                f"Omit both for open-access datasets (e.g. MIMIC-IV demo)."
            )
        self._base_url = base_url if base_url.endswith("/") else base_url + "/"
        self._unarchive = unarchive
        self._cleanup_archive = cleanup_archive
        auth = (username, password) if username is not None else None
        # Inject the Wget-prefixed default UA (see class docstring) unless the caller
        # supplied their own — header names are case-insensitive on the wire, so the
        # presence check must be too. ``headers`` is only consumed by ``_make_client``,
        # so an injected ``client=`` is untouched (its headers are the caller's).
        headers = dict(headers) if headers else {}
        if not any(k.lower() == "user-agent" for k in headers):
            headers["User-Agent"] = _default_user_agent()
        super().__init__(
            urls=None,
            client=client,
            auth=auth,
            headers=headers,
            timeout=timeout,
            max_attempts=max_attempts,
            transport=transport,
            retry_wait=retry_wait,
            include=include,
            exclude=exclude,
        )

    def _list_files(self) -> Iterable[RemoteFile]:
        sums_url = self._base_url + "SHA256SUMS.txt"
        # ``_get`` applies the source-level retry policy (5xx + transient transport
        # errors), so the manifest GET retries identically for built and injected
        # clients; 4xx comes back unwrapped and fails fast here.
        r = self._get(sums_url)
        # physionet's ``/files/`` UA gate rejects with a 403 *before* credentials are
        # considered, and — unlike a genuine auth failure — without a
        # ``WWW-Authenticate`` challenge. Surface that case legibly instead of a bare
        # ``HTTPStatusError`` indistinguishable from bad credentials. A 403 *with* a
        # challenge is a real auth failure and keeps the ordinary 4xx path below.
        if r.status_code == 403 and "WWW-Authenticate" not in r.headers:
            sent_ua = r.request.headers.get("User-Agent", "<none>")
            raise ValueError(
                f"{type(self).__name__}: got 403 with no WWW-Authenticate challenge for "
                f"{sums_url}. physionet.org serves credentialed /files/ paths only to "
                f"clients whose User-Agent starts with 'Wget/<version>' (prefix match; "
                f"anything appended after is preserved), and rejects other UAs before "
                f"credentials are considered. This client sent User-Agent: {sent_ua!r}. "
                f"If that is already Wget/-prefixed, the likely cause is missing "
                f"credentials or an unsigned data-use agreement for this dataset."
            )
        r.raise_for_status()
        for entry in self._parse_sha256sums(r.text):
            yield RemoteFile(
                rel_path=entry["rel_path"],
                sha256=entry["sha256"],
                # Percent-encode the path segment: a rel_path containing ``#``,
                # ``?``, or ``%`` would otherwise be parsed as fragment / query /
                # existing-escape and silently request the wrong resource.
                source_path=self._base_url + quote(entry["rel_path"], safe="/"),
                unarchive=self._unarchive,
                cleanup_archive=self._cleanup_archive,
            )

    @staticmethod
    def _parse_sha256sums(text: str) -> list[dict]:
        """Parse PhysioNet's ``SHA256SUMS.txt`` format.

        Lines look like:

        .. code-block:: text

            9c3a...f2  subdir/file.csv.gz
            abc1...23  README.md

        The separator is arbitrary whitespace (two spaces by convention on PhysioNet, but
        some manifests use tabs). Blank lines and comment lines (leading ``#``) are
        skipped.

        Examples:
            >>> text = (
            ...     "abc123  foo.csv\\n"
            ...     "def456  sub/bar.csv.gz\\n"
            ... )
            >>> PhysioNetSource._parse_sha256sums(text)
            [{'sha256': 'abc123', 'rel_path': 'foo.csv'}, {'sha256': 'def456', 'rel_path': 'sub/bar.csv.gz'}]

            Blank / comment lines are tolerated:

            >>> PhysioNetSource._parse_sha256sums("# header\\n\\nabc  x.csv\\n")
            [{'sha256': 'abc', 'rel_path': 'x.csv'}]

            Paths with spaces (rare on PhysioNet but legal) are preserved:

            >>> PhysioNetSource._parse_sha256sums("abc  folder with spaces/file.txt\\n")
            [{'sha256': 'abc', 'rel_path': 'folder with spaces/file.txt'}]

            Malformed lines raise:

            >>> PhysioNetSource._parse_sha256sums("no_separator_on_this_line\\n")
            Traceback (most recent call last):
                ...
            ValueError: Malformed SHA256SUMS line: 'no_separator_on_this_line'
        """
        out = []
        for raw in text.splitlines():
            line = raw.strip()
            if not line or line.startswith("#"):
                continue
            parts = line.split(None, 1)
            if len(parts) != 2:
                raise ValueError(f"Malformed SHA256SUMS line: {raw!r}")
            sha, rel = parts
            out.append({"sha256": sha, "rel_path": rel})
        return out

_parse_sha256sums(text) staticmethod

Parse PhysioNet’s SHA256SUMS.txt format.

Lines look like:

.. code-block:: text

9c3a...f2  subdir/file.csv.gz
abc1...23  README.md

The separator is arbitrary whitespace (two spaces by convention on PhysioNet, but some manifests use tabs). Blank lines and comment lines (leading #) are skipped.

Examples:

>>> text = (
...     "abc123  foo.csv\n"
...     "def456  sub/bar.csv.gz\n"
... )
>>> PhysioNetSource._parse_sha256sums(text)
[{'sha256': 'abc123', 'rel_path': 'foo.csv'}, {'sha256': 'def456', 'rel_path': 'sub/bar.csv.gz'}]

Blank / comment lines are tolerated:

>>> PhysioNetSource._parse_sha256sums("# header\n\nabc  x.csv\n")
[{'sha256': 'abc', 'rel_path': 'x.csv'}]

Paths with spaces (rare on PhysioNet but legal) are preserved:

>>> PhysioNetSource._parse_sha256sums("abc  folder with spaces/file.txt\n")
[{'sha256': 'abc', 'rel_path': 'folder with spaces/file.txt'}]

Malformed lines raise:

>>> PhysioNetSource._parse_sha256sums("no_separator_on_this_line\n")
Traceback (most recent call last):
    ...
ValueError: Malformed SHA256SUMS line: 'no_separator_on_this_line'
Source code in MEDS_extract/download/backends/physionet.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
@staticmethod
def _parse_sha256sums(text: str) -> list[dict]:
    """Parse PhysioNet's ``SHA256SUMS.txt`` format.

    Lines look like:

    .. code-block:: text

        9c3a...f2  subdir/file.csv.gz
        abc1...23  README.md

    The separator is arbitrary whitespace (two spaces by convention on PhysioNet, but
    some manifests use tabs). Blank lines and comment lines (leading ``#``) are
    skipped.

    Examples:
        >>> text = (
        ...     "abc123  foo.csv\\n"
        ...     "def456  sub/bar.csv.gz\\n"
        ... )
        >>> PhysioNetSource._parse_sha256sums(text)
        [{'sha256': 'abc123', 'rel_path': 'foo.csv'}, {'sha256': 'def456', 'rel_path': 'sub/bar.csv.gz'}]

        Blank / comment lines are tolerated:

        >>> PhysioNetSource._parse_sha256sums("# header\\n\\nabc  x.csv\\n")
        [{'sha256': 'abc', 'rel_path': 'x.csv'}]

        Paths with spaces (rare on PhysioNet but legal) are preserved:

        >>> PhysioNetSource._parse_sha256sums("abc  folder with spaces/file.txt\\n")
        [{'sha256': 'abc', 'rel_path': 'folder with spaces/file.txt'}]

        Malformed lines raise:

        >>> PhysioNetSource._parse_sha256sums("no_separator_on_this_line\\n")
        Traceback (most recent call last):
            ...
        ValueError: Malformed SHA256SUMS line: 'no_separator_on_this_line'
    """
    out = []
    for raw in text.splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        parts = line.split(None, 1)
        if len(parts) != 2:
            raise ValueError(f"Malformed SHA256SUMS line: {raw!r}")
        sha, rel = parts
        out.append({"sha256": sha, "rel_path": rel})
    return out

RemoteFile dataclass

One manifest row from a :class:Source — a frozen, self-validating POD.

Constructed inside a backend’s _list_files (in-repo backends and downstream :class:Source subclasses alike). Validation runs at construction, so a malformed row fails the instant it is built — before any filesystem or network I/O — rather than mid-orchestration.

Attributes:

Name Type Description
rel_path str

Where the file lands under download_all’s dest_dir. Must use forward slashes; path semantics mirror pathlib.PurePosixPath. Rejected at construction when absolute, containing backslashes, or escaping the destination directory after normalization.
source_path str

The source-side address as a plain string. HTTP-backed sources put the absolute URL here; fsspec-backed sources put the :class:~upath.UPath spec (which the backend re-instantiates as a UPath inside its :meth:Source._pull). Required — every real backend has somewhere to fetch from; test stubs that override _pull to write directly should pass a placeholder (the empty string is fine).
sha256 str | None

Expected SHA-256 digest (hex; normalized to lowercase at construction, rejected if not 64 hex chars). Backends that can produce one (PhysioNet from SHA256SUMS.txt, fsspec by hashing the source file, HTTP from explicit per-URL sha256: config) should set it — it’s the only verifier the orchestrator trusts to skip a re-fetch. None means “no manifest-side hash”; an existing dest can then never be proven complete, so the orchestrator re-fetches it on every run rather than trusting it.
unarchive str | None

Optional post-fetch unpack format. None (default) means no unpack. "zip", "tar", "tar.gz" / "tgz" dispatch to the matching :class:~MEDS_extract.download.unarchive.ArchiveFormat; "auto" infers the format from rel_path’s extension — useful when a single source lists both archive and non-archive files, since "auto" is a no-op on anything that doesn’t end in a recognized archive extension. Validated at construction against :class:~MEDS_extract.download.unarchive.ArchiveFormat.
cleanup_archive bool | None

Tri-state controlling whether the source archive file is removed after a successful extraction. None (default) means “use the mode-implied default”: "auto" removes the archive (the one-arg “fetch + extract + cleanup” flow); explicit formats keep it. Set True to force cleanup, False to force keep, regardless of mode. Has no effect when unarchive is None.

Examples:

Malformed rows fail at construction, not at fetch time:

>>> RemoteFile("../escape.txt", "")
Traceback (most recent call last):
    ...
ValueError: rel_path '../escape.txt' escapes dest_dir (normalizes to '../escape.txt').
>>> RemoteFile("/abs/path.txt", "")
Traceback (most recent call last):
    ...
ValueError: rel_path must be relative, got absolute: '/abs/path.txt'
>>> RemoteFile("sub/..", "")
Traceback (most recent call last):
    ...
ValueError: rel_path 'sub/..' escapes dest_dir (normalizes to '.').
>>> RemoteFile("sub\\file.txt", "")
Traceback (most recent call last):
    ...
ValueError: rel_path 'sub\\file.txt' contains backslashes; use forward slashes.
>>> RemoteFile("x.txt", "", sha256="abc123")
Traceback (most recent call last):
    ...
ValueError: sha256 must be 64 hex chars, got 'abc123'

Uppercase digests are accepted and normalized to lowercase (the compare sites hash with :func:hashlib.sha256, which emits lowercase):

>>> RemoteFile("x.txt", "", sha256="A" * 64).sha256 == "a" * 64
True

unarchive is validated at construction too — a typo’d format fails before any I/O rather than surfacing mid-download:

>>> RemoteFile("x.rar", "", unarchive="rar")
Traceback (most recent call last):
    ...
ValueError: 'rar' is not a valid ArchiveFormat
>>> RemoteFile("bundle.zip", "", unarchive="zip", cleanup_archive=True).unarchive
'zip'
Source code in MEDS_extract/download/source.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@dataclass(frozen=True)
class RemoteFile:
    """One manifest row from a :class:`Source` — a frozen, self-validating POD.

    Constructed inside a backend's ``_list_files`` (in-repo backends and downstream
    :class:`Source` subclasses alike). Validation runs at construction, so a
    malformed row fails the instant it is built — before any filesystem or network
    I/O — rather than mid-orchestration.

    Attributes:
        rel_path: Where the file lands under ``download_all``'s ``dest_dir``. Must
            use forward slashes; path semantics mirror ``pathlib.PurePosixPath``.
            Rejected at construction when absolute, containing backslashes, or
            escaping the destination directory after normalization.
        source_path: The source-side address as a plain string. HTTP-backed sources
            put the absolute URL here; fsspec-backed sources put the
            :class:`~upath.UPath` spec (which the backend re-instantiates as a
            ``UPath`` inside its :meth:`Source._pull`). Required — every real
            backend has somewhere to fetch from; test stubs that override
            ``_pull`` to write directly should pass a placeholder (the empty
            string is fine).
        sha256: Expected SHA-256 digest (hex; normalized to lowercase at
            construction, rejected if not 64 hex chars). Backends that can produce
            one (PhysioNet from ``SHA256SUMS.txt``, fsspec by hashing the source
            file, HTTP from explicit per-URL ``sha256:`` config) should set it —
            it's the only verifier the orchestrator trusts to skip a re-fetch.
            ``None`` means "no manifest-side hash"; an existing dest can then
            never be proven complete, so the orchestrator re-fetches it on every
            run rather than trusting it.
        unarchive: Optional post-fetch unpack format. ``None`` (default) means no
            unpack. ``"zip"``, ``"tar"``, ``"tar.gz"`` / ``"tgz"`` dispatch to the
            matching :class:`~MEDS_extract.download.unarchive.ArchiveFormat`;
            ``"auto"`` infers the format from ``rel_path``'s extension — useful
            when a single source lists both archive and non-archive files, since
            ``"auto"`` is a no-op on anything that doesn't end in a recognized
            archive extension. Validated at construction against
            :class:`~MEDS_extract.download.unarchive.ArchiveFormat`.
        cleanup_archive: Tri-state controlling whether the source archive file is
            removed after a successful extraction. ``None`` (default) means "use
            the mode-implied default": ``"auto"`` removes the archive (the one-arg
            "fetch + extract + cleanup" flow); explicit formats keep it. Set
            ``True`` to force cleanup, ``False`` to force keep, regardless of
            mode. Has no effect when ``unarchive`` is ``None``.

    Examples:
        Malformed rows fail at construction, not at fetch time:

        >>> RemoteFile("../escape.txt", "")
        Traceback (most recent call last):
            ...
        ValueError: rel_path '../escape.txt' escapes dest_dir (normalizes to '../escape.txt').
        >>> RemoteFile("/abs/path.txt", "")
        Traceback (most recent call last):
            ...
        ValueError: rel_path must be relative, got absolute: '/abs/path.txt'
        >>> RemoteFile("sub/..", "")
        Traceback (most recent call last):
            ...
        ValueError: rel_path 'sub/..' escapes dest_dir (normalizes to '.').
        >>> RemoteFile("sub\\\\file.txt", "")
        Traceback (most recent call last):
            ...
        ValueError: rel_path 'sub\\\\file.txt' contains backslashes; use forward slashes.
        >>> RemoteFile("x.txt", "", sha256="abc123")
        Traceback (most recent call last):
            ...
        ValueError: sha256 must be 64 hex chars, got 'abc123'

        Uppercase digests are accepted and normalized to lowercase (the compare
        sites hash with :func:`hashlib.sha256`, which emits lowercase):

        >>> RemoteFile("x.txt", "", sha256="A" * 64).sha256 == "a" * 64
        True

        ``unarchive`` is validated at construction too — a typo'd format fails
        before any I/O rather than surfacing mid-download:

        >>> RemoteFile("x.rar", "", unarchive="rar")
        Traceback (most recent call last):
            ...
        ValueError: 'rar' is not a valid ArchiveFormat
        >>> RemoteFile("bundle.zip", "", unarchive="zip", cleanup_archive=True).unarchive
        'zip'
    """

    rel_path: str
    source_path: str
    sha256: str | None = None
    unarchive: str | None = None
    cleanup_archive: bool | None = None

    def __post_init__(self):
        # rel_paths are documented as forward-slash posix paths. A backslash
        # would round-trip through ``Path(rel_path)`` differently on Windows
        # vs. POSIX, so the validation here (posixpath) would disagree with
        # ``Source._resolve_dest`` later (``Path``). Reject up-front.
        if "\\" in self.rel_path:
            raise ValueError(f"rel_path {self.rel_path!r} contains backslashes; use forward slashes.")
        if posixpath.isabs(self.rel_path):
            raise ValueError(f"rel_path must be relative, got absolute: {self.rel_path!r}")
        norm = posixpath.normpath(self.rel_path)
        if norm in (".", "..") or norm.startswith("../"):
            raise ValueError(f"rel_path {self.rel_path!r} escapes dest_dir (normalizes to {norm!r}).")
        if self.sha256 is not None:
            if not _SHA256_RE.fullmatch(self.sha256):
                raise ValueError(f"sha256 must be 64 hex chars, got {self.sha256!r}")
            object.__setattr__(self, "sha256", self.sha256.lower())
        if self.unarchive is not None:
            ArchiveFormat(self.unarchive)  # raises ValueError on unknown tokens

    @property
    def dest_key(self) -> str:
        """The posix-normalized ``rel_path`` — the collision key under a shared dest_dir.

        Two rows whose ``dest_key`` matches would race on the same ``.part`` file
        under concurrent workers, so both :attr:`Source.files` (within one source)
        and :func:`validate_unique_destinations` (across sources) reject them.

        Examples:
            >>> RemoteFile("sub/../a.txt", "").dest_key
            'a.txt'
        """
        return posixpath.normpath(self.rel_path)

dest_key property

The posix-normalized rel_path — the collision key under a shared dest_dir.

Two rows whose dest_key matches would race on the same .part file under concurrent workers, so both :attr:Source.files (within one source) and :func:validate_unique_destinations (across sources) reject them.

Examples:

>>> RemoteFile("sub/../a.txt", "").dest_key
'a.txt'

Source

Bases: ABC

A place raw data comes from.

Subclasses implement two private hooks:

  • :meth:_list_files — enumerate what files the source offers as :class:RemoteFile rows.
  • :meth:_pull — stream the bytes at one source address into a target path.

The base class supplies the public surface — :meth:download_all for the bundle, :attr:files for the validated manifest — plus all the cross-cutting behavior every backend needs: .part staging, SHA-256 verification, atomic rename, path-traversal validation, duplicate-destination detection, include/exclude manifest filtering, and the sequential / parallel orchestration.

Parameters:

Name Type Description Default
include list[str] | None

Optional list of :mod:fnmatch-style globs. When set, only manifest rows whose normalized rel_path matches at least one pattern are downloaded. None (default) selects everything; an empty list matches nothing (standard fnmatch semantics), so include=[] selects zero files.

 None
exclude list[str] | None

Optional list of :mod:fnmatch-style globs. Rows matching any pattern are dropped (applied after include). None (default) and [] both drop nothing.

 None

Invariants subclasses must uphold:

  • :meth:_list_files is idempotent across calls — re-enumerating must produce the same set of :class:RemoteFile rows (in the same order when possible).
  • :meth:_pull writes the bytes at source_path into target and raises on any transport error. Backends with resume semantics (e.g. HTTP Range) MAY inspect existing content at target and append; backends without resume should overwrite.
  • Subclasses that define __init__ should call super().__init__(...) to wire the include / exclude filters through.

Concrete usage examples live on the methods that implement them: :meth:download_all (the public entry + orchestration policy), :attr:files (manifest validation + filtering), :meth:_fetch_one (the per-file pipeline: skip/re-fetch policy + staging + verify + rename).

Source code in MEDS_extract/download/source.py 
 228
 229
 230
 231
 232
 233
 234
 235
 236
 237
 238
 239
 240
 241
 242
 243
 244
 245
 246
 247
 248
 249
 250
 251
 252
 253
 254
 255
 256
 257
 258
 259
 260
 261
 262
 263
 264
 265
 266
 267
 268
 269
 270
 271
 272
 273
 274
 275
 276
 277
 278
 279
 280
 281
 282
 283
 284
 285
 286
 287
 288
 289
 290
 291
 292
 293
 294
 295
 296
 297
 298
 299
 300
 301
 302
 303
 304
 305
 306
 307
 308
 309
 310
 311
 312
 313
 314
 315
 316
 317
 318
 319
 320
 321
 322
 323
 324
 325
 326
 327
 328
 329
 330
 331
 332
 333
 334
 335
 336
 337
 338
 339
 340
 341
 342
 343
 344
 345
 346
 347
 348
 349
 350
 351
 352
 353
 354
 355
 356
 357
 358
 359
 360
 361
 362
 363
 364
 365
 366
 367
 368
 369
 370
 371
 372
 373
 374
 375
 376
 377
 378
 379
 380
 381
 382
 383
 384
 385
 386
 387
 388
 389
 390
 391
 392
 393
 394
 395
 396
 397
 398
 399
 400
 401
 402
 403
 404
 405
 406
 407
 408
 409
 410
 411
 412
 413
 414
 415
 416
 417
 418
 419
 420
 421
 422
 423
 424
 425
 426
 427
 428
 429
 430
 431
 432
 433
 434
 435
 436
 437
 438
 439
 440
 441
 442
 443
 444
 445
 446
 447
 448
 449
 450
 451
 452
 453
 454
 455
 456
 457
 458
 459
 460
 461
 462
 463
 464
 465
 466
 467
 468
 469
 470
 471
 472
 473
 474
 475
 476
 477
 478
 479
 480
 481
 482
 483
 484
 485
 486
 487
 488
 489
 490
 491
 492
 493
 494
 495
 496
 497
 498
 499
 500
 501
 502
 503
 504
 505
 506
 507
 508
 509
 510
 511
 512
 513
 514
 515
 516
 517
 518
 519
 520
 521
 522
 523
 524
 525
 526
 527
 528
 529
 530
 531
 532
 533
 534
 535
 536
 537
 538
 539
 540
 541
 542
 543
 544
 545
 546
 547
 548
 549
 550
 551
 552
 553
 554
 555
 556
 557
 558
 559
 560
 561
 562
 563
 564
 565
 566
 567
 568
 569
 570
 571
 572
 573
 574
 575
 576
 577
 578
 579
 580
 581
 582
 583
 584
 585
 586
 587
 588
 589
 590
 591
 592
 593
 594
 595
 596
 597
 598
 599
 600
 601
 602
 603
 604
 605
 606
 607
 608
 609
 610
 611
 612
 613
 614
 615
 616
 617
 618
 619
 620
 621
 622
 623
 624
 625
 626
 627
 628
 629
 630
 631
 632
 633
 634
 635
 636
 637
 638
 639
 640
 641
 642
 643
 644
 645
 646
 647
 648
 649
 650
 651
 652
 653
 654
 655
 656
 657
 658
 659
 660
 661
 662
 663
 664
 665
 666
 667
 668
 669
 670
 671
 672
 673
 674
 675
 676
 677
 678
 679
 680
 681
 682
 683
 684
 685
 686
 687
 688
 689
 690
 691
 692
 693
 694
 695
 696
 697
 698
 699
 700
 701
 702
 703
 704
 705
 706
 707
 708
 709
 710
 711
 712
 713
 714
 715
 716
 717
 718
 719
 720
 721
 722
 723
 724
 725
 726
 727
 728
 729
 730
 731
 732
 733
 734
 735
 736
 737
 738
 739
 740
 741
 742
 743
 744
 745
 746
 747
 748
 749
 750
 751
 752
 753
 754
 755
 756
 757
 758
 759
 760
 761
 762
 763
 764
 765
 766
 767
 768
 769
 770
 771
 772
 773
 774
 775
 776
 777
 778
 779
 780
 781
 782
 783
 784
 785
 786
 787
 788
 789
 790
 791
 792
 793
 794
 795
 796
 797
 798
 799
 800
 801
 802
 803
 804
 805
 806
 807
 808
 809
 810
 811
 812
 813
 814
 815
 816
 817
 818
 819
 820
 821
 822
 823
 824
 825
 826
 827
 828
 829
 830
 831
 832
 833
 834
 835
 836
 837
 838
 839
 840
 841
 842
 843
 844
 845
 846
 847
 848
 849
 850
 851
 852
 853
 854
 855
 856
 857
 858
 859
 860
 861
 862
 863
 864
 865
 866
 867
 868
 869
 870
 871
 872
 873
 874
 875
 876
 877
 878
 879
 880
 881
 882
 883
 884
 885
 886
 887
 888
 889
 890
 891
 892
 893
 894
 895
 896
 897
 898
 899
 900
 901
 902
 903
 904
 905
 906
 907
 908
 909
 910
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
class Source(ABC):
    """A place raw data comes from.

    Subclasses implement two private hooks:

    - :meth:`_list_files` — enumerate what files the source offers as
      :class:`RemoteFile` rows.
    - :meth:`_pull` — stream the bytes at one source address into a target path.

    The base class supplies the public surface — :meth:`download_all` for the
    bundle, :attr:`files` for the validated manifest — plus all the cross-cutting
    behavior every backend needs: ``.part`` staging, SHA-256 verification, atomic
    rename, path-traversal validation, duplicate-destination detection,
    include/exclude manifest filtering, and the sequential / parallel
    orchestration.

    Args:
        include: Optional list of :mod:`fnmatch`-style globs. When set, only
            manifest rows whose normalized ``rel_path`` matches at least one
            pattern are downloaded. ``None`` (default) selects everything; an
            empty list matches nothing (standard fnmatch semantics), so
            ``include=[]`` selects zero files.
        exclude: Optional list of :mod:`fnmatch`-style globs. Rows matching any
            pattern are dropped (applied after ``include``). ``None`` (default)
            and ``[]`` both drop nothing.

    Invariants subclasses must uphold:

    - :meth:`_list_files` is idempotent across calls — re-enumerating must produce
      the same set of :class:`RemoteFile` rows (in the same order when possible).
    - :meth:`_pull` writes the bytes at ``source_path`` into ``target`` and
      raises on any transport error. Backends with resume semantics (e.g. HTTP
      ``Range``) MAY inspect existing content at ``target`` and append;
      backends without resume should overwrite.
    - Subclasses that define ``__init__`` should call ``super().__init__(...)``
      to wire the ``include`` / ``exclude`` filters through.

    Concrete usage examples live on the methods that implement them:
    :meth:`download_all` (the public entry + orchestration policy),
    :attr:`files` (manifest validation + filtering), :meth:`_fetch_one` (the
    per-file pipeline: skip/re-fetch policy + staging + verify + rename).
    """

    # Class-level fallbacks so subclasses that define ``__init__`` without calling
    # ``super().__init__`` still get well-defined (unfiltered) behavior.
    _include: list[str] | None = None
    _exclude: list[str] | None = None

    def __init__(self, include: list[str] | None = None, exclude: list[str] | None = None):
        # ``is not None`` (not truthiness): ``include=[]`` must mean "no pattern
        # matches anything" — i.e. select zero files — per fnmatch semantics, not
        # silently collapse to "select the entire release".
        self._include = list(include) if include is not None else None
        self._exclude = list(exclude) if exclude is not None else None

    def download_all(
        self,
        dest_dir: str | Path,
        *,
        pool: Executor | None = None,
        continue_on_error: bool = False,
        do_overwrite: bool = False,
    ) -> None:
        """Download every file this source lists into ``dest_dir``.

        Args:
            dest_dir: Where files land. Created if missing.
            pool: Optional :class:`~concurrent.futures.Executor` (typically a
                :class:`~concurrent.futures.ThreadPoolExecutor`) to submit work
                to. The caller owns the pool's lifetime. When ``None`` (default),
                the bundle is fetched sequentially in the calling thread — no
                thread pool is created. Pass a pool when you want parallelism,
                sized to whatever your transport tolerates.
            continue_on_error: If ``False`` (default), the first per-file failure
                propagates. If ``True``, per-file errors are collected and raised
                as a single :class:`ExceptionGroup` at the end so the caller sees
                every failure, not just the first.
            do_overwrite: If ``True``, skip the verified-dest check and clear
                ``dest`` / ``.part`` before each fetch — re-fetches everything
                from scratch, even files whose local copy verifies.

        Raises:
            Exception: From the transport layer on any per-file failure when
                ``continue_on_error=False``.
            ExceptionGroup: When ``continue_on_error=True`` and at least one
                file failed.
            ValueError: When the manifest contains an unsafe rel_path (raised at
                :class:`RemoteFile` construction) or duplicate destinations
                (raised by :attr:`files`).

        Examples:
            Simple case — no pool passed, ``download_all`` runs sequentially:

            >>> class StubSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("a.txt", ""), RemoteFile("sub/b.txt", "")]
            ...     def _pull(self, source_path, target):
            ...         target.write_text(f"contents of {target.name}")
            >>>
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     StubSource().download_all(d)
            ...     print_directory(d)
            ├── a.txt
            └── sub
                └── b.txt

            Multi-source case — caller owns one :class:`ThreadPoolExecutor` and
            hands it to every source. Two sources writing distinct files into one
            ``dest_dir`` is the typical CLI pattern (one ``physionet`` source plus
            one ``http`` source for a metadata bundle):

            >>> from concurrent.futures import ThreadPoolExecutor
            >>>
            >>> class SourceA(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("a.txt", "")]
            ...     def _pull(self, source_path, target):
            ...         target.write_text("from A")
            >>>
            >>> class SourceB(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("metadata/b.csv", "")]
            ...     def _pull(self, source_path, target):
            ...         target.write_text("from B")
            >>>
            >>> with tempfile.TemporaryDirectory() as d, ThreadPoolExecutor(max_workers=4) as pool:
            ...     d = Path(d)
            ...     for src in [SourceA(), SourceB()]:
            ...         src.download_all(d, pool=pool)
            ...     print_directory(d)
            ├── a.txt
            └── metadata
                └── b.csv

            Already-complete files are skipped — ``_pull`` is not invoked for any
            :class:`RemoteFile` whose on-disk copy verifies against the manifest's
            ``sha256``:

            >>> import hashlib
            >>> body = b"abc"
            >>> digest = hashlib.sha256(body).hexdigest()
            >>>
            >>> class SkipSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("x.txt", "", sha256=digest)]
            ...     def _pull(self, source_path, target):
            ...         raise RuntimeError("must not be called — file is already complete")
            >>>
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     _ = (d / "x.txt").write_bytes(body)
            ...     SkipSource().download_all(d)  # no exception → already-complete skip worked

            ``do_overwrite=True`` re-fetches even a file whose on-disk copy
            verifies — ``_pull`` runs exactly once across the two calls below
            (skipped without overwrite, forced with it):

            >>> pulls = []
            >>> class CountingSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("x.txt", "", sha256=digest)]
            ...     def _pull(self, source_path, target):
            ...         pulls.append(source_path)
            ...         target.write_bytes(body)
            >>>
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     _ = (d / "x.txt").write_bytes(body)
            ...     CountingSource().download_all(d)  # verified on disk → skipped
            ...     CountingSource().download_all(d, do_overwrite=True)  # forced re-fetch
            ...     len(pulls)
            1

            An existing ``dest`` with **no manifest sha** can never be proven
            complete, so it is re-fetched rather than trusted — the stale local
            copy is replaced atomically, and re-runs stay idempotent:

            >>> class UnverifiableSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("x.txt", "")]  # no sha
            ...     def _pull(self, source_path, target):
            ...         target.write_text("fresh")
            >>>
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     _ = (d / "x.txt").write_bytes(b"stale")
            ...     UnverifiableSource().download_all(d)
            ...     UnverifiableSource().download_all(d)  # re-run: re-fetches again, no error
            ...     print((d / "x.txt").read_text())
            fresh

            An existing ``dest`` whose content **mismatches** the manifest sha is
            likewise re-fetched (with a warning naming the file) — and the fresh
            bytes must still verify before the atomic replace:

            >>> class MismatchRepairSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("x.txt", "", sha256=digest)]
            ...     def _pull(self, source_path, target):
            ...         target.write_bytes(body)
            >>>
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     _ = (d / "x.txt").write_bytes(b"corrupt local copy")
            ...     MismatchRepairSource().download_all(d)
            ...     print((d / "x.txt").read_bytes() == body)
            True

            Failure policy: by default the first per-file failure propagates and
            later files are not attempted; ``continue_on_error=True`` attempts
            everything and collects the failures into one :class:`ExceptionGroup`:

            >>> class FlakySource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("bad.txt", "bad"), RemoteFile("good.txt", "ok")]
            ...     def _pull(self, source_path, target):
            ...         if source_path == "bad":
            ...             raise RuntimeError("transport boom")
            ...         target.write_text("ok")
            >>>
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     try:
            ...         FlakySource().download_all(d)
            ...     except RuntimeError as e:
            ...         print(f"raised: {e}; good.txt fetched: {(d / 'good.txt').exists()}")
            raised: transport boom; good.txt fetched: False

            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     try:
            ...         FlakySource().download_all(d, continue_on_error=True)
            ...     except ExceptionGroup as eg:
            ...         print(f"{len(eg.exceptions)} failed; good.txt fetched: {(d / 'good.txt').exists()}")
            1 failed; good.txt fetched: True

            Path-traversal manifests and absolute paths are rejected at
            :class:`RemoteFile` construction; duplicate destinations are rejected
            at :attr:`files` (the first thing ``download_all`` accesses) — see
            those docstrings for examples.
        """
        # Materialize + validate the manifest before touching the filesystem, so a
        # malformed ``sources:`` entry doesn't leave behind an empty ``dest_dir``.
        items = self.files
        dest_dir = Path(dest_dir)
        dest_dir.mkdir(parents=True, exist_ok=True)
        logger.info(f"Fetching {self.n_files} files to {dest_dir} ({'pooled' if pool else 'sequential'})")

        errors: list[Exception] = []
        counts = {"fetched": 0, "skipped": 0, "promoted": 0}
        n_failed = 0
        total_bytes = 0
        fetched_bytes = 0
        t0 = last_progress = time.monotonic()
        # ``closing`` guarantees the generator's ``finally`` runs even when the loop
        # exits early via ``raise`` (fail-fast) — in pooled mode that ``finally`` is
        # what cancels the still-queued futures so "fail fast" actually stops the run.
        # Tripped the moment a fetch fails, so queued work stops before the unwind
        # reaches the generator's teardown. See ``_attempts``.
        abort = threading.Event()
        attempts = self._attempts(self._iter_attempts(items, dest_dir, do_overwrite), pool, abort)
        try:
            with closing(attempts):
                for item, run in attempts:
                    try:
                        status, n_bytes = run()
                    except Exception as e:
                        # Tag the exception with the item it came from so a caller
                        # inspecting the ExceptionGroup (or a bare re-raise) can tell
                        # which file failed without cross-referencing logs.
                        e.add_note(f"while fetching {item.rel_path!r} from {item.source_path!r}")
                        n_failed += 1
                        if not continue_on_error:
                            abort.set()
                            raise
                        logger.exception(f"Failed to fetch {item.rel_path}")
                        errors.append(e)
                    else:
                        counts[status] += 1
                        total_bytes += n_bytes
                        if status == "fetched":
                            fetched_bytes += n_bytes
                    now = time.monotonic()
                    if now - last_progress >= _PROGRESS_INTERVAL_S:
                        n_done = sum(counts.values()) + n_failed
                        logger.info(
                            f"Progress: {n_done}/{self.n_files} files "
                            f"({total_bytes / 2**20:.0f} MiB) in {now - t0:.0f}s "
                            f"({total_bytes / 2**20 / max(now - t0, 1e-9):.1f} MiB/s)"
                        )
                        last_progress = now
        finally:
            # Emitted in a ``finally`` so a fail-fast exit still reports the partial
            # totals a multi-hour run accumulated before the failure.
            elapsed = time.monotonic() - t0
            fetched_mib = fetched_bytes / 2**20
            logger.info(
                f"{type(self).__name__}: {counts['fetched']} fetched "
                f"({fetched_mib:.1f} MiB in {elapsed:.1f}s, "
                f"{fetched_mib / max(elapsed, 1e-9):.1f} MiB/s), "
                f"{counts['skipped']} skipped, {counts['promoted']} promoted, "
                f"{n_failed} failed of {self.n_files} files -> {dest_dir}"
            )
        if errors:
            raise ExceptionGroup(f"{len(errors)} of {self.n_files} files failed to download", errors)

    @cached_property
    def files(self) -> list[RemoteFile]:
        """The validated manifest — calls :meth:`_list_files` once, materializes, filters, and validates.

        Cached on first access. Subsequent ``download_all`` calls reuse the same
        list rather than re-hitting :meth:`_list_files` (which may do network
        I/O — e.g. PhysioNet fetches ``SHA256SUMS.txt``). If a source's contents
        could change between runs and the caller wants a fresh manifest, build
        a new ``Source`` instance.

        Per-row validation (relative, no traversal, no backslashes, well-formed
        sha256) happens at :class:`RemoteFile` construction inside
        :meth:`_list_files`, so it needs no re-checking here. This property adds
        the two whole-manifest steps:

        - **include / exclude filtering** — the constructor's glob patterns are
          matched against each row's normalized ``rel_path``; rows an ``include``
          list doesn't match, or an ``exclude`` list does match, are dropped.
        - **duplicate-destination detection** — two rows whose normalized
          rel_paths collide (``a/../x.csv`` vs ``x.csv``) would race on the same
          ``.part`` file under concurrent workers, so they fail the bundle
          up-front.

        ``_fetch_one`` calls :meth:`_resolve_dest` per-item at fetch time as the
        runtime security boundary (e.g. for dest_dirs that contain symlinks).

        Examples:
            Duplicate destinations are caught even when the strings differ:

            >>> class DupSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("a.txt", ""), RemoteFile("sub/../a.txt", "")]
            ...     def _pull(self, source_path, target):
            ...         target.write_text("never reached")
            >>>
            >>> DupSource().files
            Traceback (most recent call last):
                ...
            ValueError: Duplicate destination 'sub/../a.txt': collides with 'a.txt'. ...

            Unsafe rel_paths fail earlier still — at :class:`RemoteFile`
            construction inside ``_list_files``:

            >>> class EscapingSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("../escape.txt", "")]
            ...     def _pull(self, source_path, target):
            ...         target.write_text("never reached")
            >>>
            >>> EscapingSource().files
            Traceback (most recent call last):
                ...
            ValueError: rel_path '../escape.txt' escapes dest_dir ...

            ``include`` / ``exclude`` globs subset the manifest:

            >>> class TreeSource(Source):
            ...     def _list_files(self):
            ...         return [
            ...             RemoteFile("hosp/patients.csv.gz", ""),
            ...             RemoteFile("hosp/labevents.csv.gz", ""),
            ...             RemoteFile("note/discharge.csv.gz", ""),
            ...         ]
            ...     def _pull(self, source_path, target):
            ...         target.write_text("ok")
            >>>
            >>> [f.rel_path for f in TreeSource(include=["hosp/*"]).files]
            ['hosp/patients.csv.gz', 'hosp/labevents.csv.gz']
            >>> [f.rel_path for f in TreeSource(exclude=["*/labevents*"]).files]
            ['hosp/patients.csv.gz', 'note/discharge.csv.gz']

            ``exclude`` is applied after ``include`` — both together intersect:

            >>> [f.rel_path for f in TreeSource(include=["hosp/*"], exclude=["*/labevents*"]).files]
            ['hosp/patients.csv.gz']

            An empty ``include`` list matches nothing (fnmatch semantics) — it
            selects zero files rather than silently selecting everything:

            >>> TreeSource(include=[]).files
            []
        """
        t0 = time.monotonic()
        items = list(self._list_files())
        logger.info(
            f"{type(self).__name__}: manifest listed {len(items)} files in {time.monotonic() - t0:.1f}s"
        )
        if self._include is not None or self._exclude is not None:
            kept = [item for item in items if self._selected(item)]
            # Only log when this pass actually dropped rows — backends that
            # pre-filter inside ``_list_files`` (e.g. FsspecSource, to skip
            # hashing excluded bytes) hand us an already-filtered manifest, and
            # an unconditional "kept N/N" line would misread as "filters were a
            # no-op". Those backends log their own pre-filter counts.
            if len(kept) != len(items):
                logger.info(
                    f"include/exclude filters dropped {len(items) - len(kept)} of "
                    f"{len(items)} manifest rows ({len(kept)} kept)"
                )
            items = kept
        seen: dict[str, RemoteFile] = {}
        for item in items:
            if item.dest_key in seen:
                raise ValueError(
                    f"Duplicate destination {item.rel_path!r}: collides with "
                    f"{seen[item.dest_key].rel_path!r}. Each item from a source's "
                    "_list_files() must resolve to a unique rel_path."
                )
            seen[item.dest_key] = item
        return items

    @property
    def n_files(self) -> int:
        """Number of files in the validated, filtered manifest."""
        return len(self.files)

    def _selected(self, item: RemoteFile) -> bool:
        """Apply the constructor's ``include`` / ``exclude`` globs to one manifest row."""
        return self._selected_path(item.dest_key)

    def _selected_path(self, dest_key: str) -> bool:
        """String-level filter check, for backends that want to skip expensive per-file work (e.g. hashing) on
        rows the manifest filters would drop anyway."""
        if self._include is not None and not any(fnmatch.fnmatchcase(dest_key, p) for p in self._include):
            return False
        return not (
            self._exclude is not None and any(fnmatch.fnmatchcase(dest_key, p) for p in self._exclude)
        )

    @abstractmethod
    def _list_files(self) -> Iterable[RemoteFile]:
        """Subclass hook — enumerate the files this source offers.

        :attr:`files` is the validating cached wrapper that callers use; this
        hook just produces the rows.
        """

    @abstractmethod
    def _pull(self, source_path: str, target: Path) -> None:
        """Stream the bytes at ``source_path`` into ``target``.

        ``source_path`` is whatever the backend stored in
        ``RemoteFile.source_path`` when it built the manifest (a URL for HTTP,
        a UPath spec for fsspec). On successful return ``target`` contains
        the complete file; on any transport error, raise.

        Backends with resume semantics (HTTP ``Range``) MAY observe existing
        bytes at ``target`` and append; backends without resume should
        overwrite.
        """

    def close(self) -> None:  # noqa: B027 — intentional no-op default; subclasses override when needed
        """Release transport resources held by this source.

        Default is a no-op. Subclasses that own network clients / file handles / connection pools override
        this. Safe to call multiple times; safe to call on sources that own nothing.
        """

    def __enter__(self) -> Source:
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        self.close()

    @staticmethod
    def _resolve_dest(dest_dir: Path, rel_path: str) -> Path:
        """Resolve ``rel_path`` under ``dest_dir``, rejecting any escape attempts.

        :class:`RemoteFile` construction already rejects malformed rel_paths by
        string inspection; this fetch-time check is the runtime security boundary
        against escapes that only materialize on a real filesystem (e.g. symlinks
        inside ``dest_dir``).
        """
        if Path(rel_path).is_absolute():
            raise ValueError(f"rel_path must be relative, got absolute: {rel_path!r}")
        resolved, contained = resolve_contained(dest_dir, rel_path)
        if not contained:
            raise ValueError(
                f"rel_path {rel_path!r} escapes dest_dir {Path(dest_dir).resolve()} (resolved to {resolved})."
            )
        return resolved

    @staticmethod
    def _verifies(dest: Path, item: RemoteFile) -> bool:
        """True iff ``dest`` exists AND the manifest's ``sha256`` matches.

        SHA-256 is the only verifier we trust. Same-size files can have different
        content; existence-with-no-hash means the file on disk could be anything.
        Backends that want skip-on-rerun semantics must populate ``sha256``.
        """
        return item.sha256 is not None and dest.exists() and sha256_of(dest) == item.sha256

    @staticmethod
    def _maybe_unarchive(item: RemoteFile, dest: Path) -> None:
        """Post-fetch unpack hook — runs after bytes newly land at ``dest``.

        A no-op unless ``item.unarchive`` is set. ``"auto"`` resolves the format
        from ``dest``'s extension (and is a no-op on non-archive extensions like
        ``.csv.gz``); explicit formats dispatch directly. Extraction lands in
        ``dest``'s directory via
        :func:`~MEDS_extract.download.unarchive.safe_extract`, which validates
        every member against zip-slip / tar-slip before any bytes are written.

        Tri-state cleanup: ``cleanup_archive=None`` defers to the unarchive mode —
        :attr:`~MEDS_extract.download.unarchive.ArchiveFormat.AUTO` removes the
        archive (the one-arg "fetch + extract + drop" flow), explicit formats keep
        it. Explicit ``True`` / ``False`` always wins.

        Invoked from :meth:`_fetch_one` on the ``"fetched"`` and ``"promoted"``
        paths only — a ``"skipped"`` dest was not newly written, so it is not
        re-extracted.

        Examples:
            ``unarchive="auto"`` unpacks a zip next to itself and (by AUTO's
            cleanup default) removes the archive afterwards:

            >>> import zipfile
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     archive = d / "bundle.zip"
            ...     with zipfile.ZipFile(archive, "w") as zf:
            ...         zf.writestr("sub/a.csv", "col\\n1")
            ...     Source._maybe_unarchive(RemoteFile("bundle.zip", "", unarchive="auto"), archive)
            ...     print_directory(d)
            └── sub
                └── a.csv

            An explicit format keeps the archive by default:

            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     archive = d / "bundle.zip"
            ...     with zipfile.ZipFile(archive, "w") as zf:
            ...         zf.writestr("a.csv", "col\\n1")
            ...     Source._maybe_unarchive(RemoteFile("bundle.zip", "", unarchive="zip"), archive)
            ...     print_directory(d)
            ├── a.csv
            └── bundle.zip

            ``cleanup_archive`` overrides the mode default in either direction:

            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     archive = d / "bundle.zip"
            ...     with zipfile.ZipFile(archive, "w") as zf:
            ...         zf.writestr("a.csv", "col\\n1")
            ...     item = RemoteFile("bundle.zip", "", unarchive="zip", cleanup_archive=True)
            ...     Source._maybe_unarchive(item, archive)
            ...     print_directory(d)
            └── a.csv

            ``"auto"`` on a non-archive is a no-op — the file is left alone:

            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     f = d / "patients.csv.gz"
            ...     _ = f.write_bytes(b"not an archive")
            ...     Source._maybe_unarchive(RemoteFile("patients.csv.gz", "", unarchive="auto"), f)
            ...     print_directory(d)
            └── patients.csv.gz
        """
        if not item.unarchive:
            return
        fmt = resolve_format(item.unarchive, dest)
        if fmt is None:
            return
        t0 = time.monotonic()
        safe_extract(dest, dest.parent, fmt)
        logger.debug(f"Extracted {item.rel_path} ({fmt.value}) in {time.monotonic() - t0:.1f}s")
        if item.cleanup_archive is None:
            cleanup = ArchiveFormat(item.unarchive) is ArchiveFormat.AUTO
        else:
            cleanup = item.cleanup_archive
        if cleanup:
            dest.unlink()

    def _fetch_one(self, item: RemoteFile, dest_dir: Path, do_overwrite: bool) -> tuple[str, int]:
        """Fetch one manifest entry end-to-end: policy → ``.part`` staging → verify → rename.

        Pipeline:

        1. Resolve ``dest = dest_dir / item.rel_path`` (with traversal validation).
        2. If ``do_overwrite=True``: unconditionally clear ``dest`` and any stale
           ``.part`` (whether or not ``dest`` exists), then proceed to step 5.
           Otherwise, on a pre-existing ``dest``:

           - ``dest`` verifies against ``item.sha256``: skip and return.
           - ``item.sha256`` is set but ``dest`` mismatches: re-fetch (with a
             warning naming the file) — ``dest`` is replaced only by the atomic
             rename in step 7, after the fresh bytes verify.
           - no ``item.sha256``: re-fetch — a file we can't prove matches the
             manifest is never trusted, and never skipped.

        3. If a prior run left a ``.part`` that already verifies against
           ``item.sha256`` (interrupted between the last byte and the rename),
           promote it to ``dest`` directly — no re-fetch.
        4. If the manifest has no SHA to verify against, discard any stale
           ``.part`` — resume-without-verification is unsafe.
        5. Call ``self._pull(item.source_path, part)`` — backend streams bytes.
        6. If ``item.sha256`` is set, hash ``part`` once via :func:`sha256_of`
           and compare; on mismatch, unlink ``part`` and raise
           :class:`ChecksumError`.
        7. Atomic-rename ``part`` → ``dest``.
        8. If ``item.unarchive`` is set, run the post-fetch unpack hook
           (:meth:`_maybe_unarchive`) — also applied on the promote path in
           step 3, but never to a step-2 skip (the dest was not newly written).

        Returns:
            A ``(status, n_bytes)`` tuple where ``status`` is ``"skipped"``
            (already-complete dest), ``"promoted"`` (complete ``.part``
            renamed without re-fetching), or ``"fetched"`` (bytes actually
            transferred), and ``n_bytes`` is the size of the file the status
            applies to. ``download_all`` tallies these into its end-of-bundle
            summary.

        On any exception, no new ``dest`` is created and an existing ``dest``
        is never modified — the re-fetch paths replace it only via the atomic
        rename, after the fresh bytes verify. ``part`` may exist after a
        partial transport failure (intentional — gives a future run a head
        start via Range-resume on backends that support it).

        Examples:
            Backend's ``_pull`` produces the bytes; this method handles staging,
            sha verification, and atomic rename:

            >>> import hashlib
            >>> class FakeSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("x.txt", "dummy", sha256=hashlib.sha256(b"hi").hexdigest())]
            ...     def _pull(self, source_path, target):
            ...         target.write_bytes(b"hi")
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     src = FakeSource()
            ...     [item] = src.files
            ...     src._fetch_one(item, d, do_overwrite=False)
            ...     print((d / "x.txt").read_bytes(), (d / "x.txt.part").exists())
            ('fetched', 2)
            b'hi' False

            On a SHA mismatch the staged ``.part`` is deleted, ``dest`` is not
            created, and :class:`ChecksumError` propagates:

            >>> class WrongShaSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("x.txt", "dummy", sha256="0" * 64)]
            ...     def _pull(self, source_path, target):
            ...         target.write_bytes(b"hi")
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     src = WrongShaSource()
            ...     [item] = src.files
            ...     try:
            ...         src._fetch_one(item, d, do_overwrite=False)
            ...     except ChecksumError:
            ...         print(f"raised; dest={(d / 'x.txt').exists()}, part={(d / 'x.txt.part').exists()}")
            raised; dest=False, part=False

            When the manifest has no SHA, a stale ``.part`` from a prior failed
            run can't be safely resumed (nothing would catch silent corruption),
            so it is discarded before ``_pull`` runs — the backend starts fresh:

            >>> class NoShaSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("x.txt", "dummy")]  # no sha
            ...     def _pull(self, source_path, target):
            ...         print(f"stale .part visible to _pull: {target.exists()}")
            ...         target.write_text("fresh")
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     _ = (d / "x.txt.part").write_bytes(b"stale partial")
            ...     src = NoShaSource()
            ...     [item] = src.files
            ...     src._fetch_one(item, d, do_overwrite=False)
            stale .part visible to _pull: False
            ('fetched', 5)

            A leftover ``.part`` that already verifies against the manifest sha
            (prior run died between the last byte and the rename) is promoted to
            ``dest`` directly — ``_pull`` is never invoked:

            >>> body = b"the whole file, fully written"
            >>> class NoRefetchSource(Source):
            ...     def _list_files(self):
            ...         return [RemoteFile("x.txt", "dummy", sha256=hashlib.sha256(body).hexdigest())]
            ...     def _pull(self, source_path, target):
            ...         raise AssertionError("must not re-fetch a complete .part")
            >>> with tempfile.TemporaryDirectory() as d:
            ...     d = Path(d)
            ...     _ = (d / "x.txt.part").write_bytes(body)
            ...     src = NoRefetchSource()
            ...     [item] = src.files
            ...     src._fetch_one(item, d, do_overwrite=False)
            ...     print((d / "x.txt").read_bytes() == body, (d / "x.txt.part").exists())
            ('promoted', 29)
            True False
        """
        dest = self._resolve_dest(dest_dir, item.rel_path)
        dest.parent.mkdir(parents=True, exist_ok=True)
        part = dest.with_name(dest.name + ".part")

        if do_overwrite:
            # Clear both independently — a ``.part`` from a half-finished prior
            # run can exist even when ``dest`` doesn't, and either one left in
            # place would be picked up as a Range-resume base.
            if dest.exists():
                dest.unlink()
            if part.exists():
                part.unlink()
        elif dest.exists():
            if self._verifies(dest, item):
                logger.debug(f"Skipping {item.rel_path}: already complete.")
                return ("skipped", dest.stat().st_size)
            # An existing dest that can't be proven complete is never trusted and
            # never skipped: re-fetch it. The stale copy stays in place until the
            # fresh bytes verify — the atomic rename below is what replaces it.
            if item.sha256 is None:
                logger.debug(f"Re-fetching {item.rel_path}: existing file has no manifest sha to verify.")
            else:
                logger.warning(
                    f"Re-fetching {item.rel_path}: existing file at {dest} failed SHA-256 "
                    "verification against the manifest."
                )

        if item.sha256 is not None:
            # A prior run may have died between writing the last byte and the
            # rename below — in that case the ``.part`` is the complete file and
            # re-fetching it (or bouncing off an unsatisfiable Range request)
            # wastes the whole transfer. Verify and promote directly.
            if part.exists() and sha256_of(part) == item.sha256:
                logger.debug(f"Promoting complete .part for {item.rel_path} without re-fetching.")
                n_bytes = part.stat().st_size
                part.replace(dest)
                self._maybe_unarchive(item, dest)
                return ("promoted", n_bytes)
        elif part.exists():
            # Resume-without-verification is unsafe: without a sha to catch silent
            # corruption, a stale ``.part`` could be from a different version of
            # the source file. Clear it so ``_pull`` starts fresh. With sha set,
            # Range-resume is safe because the post-write verify catches mismatches.
            part.unlink()

        t_pull = time.monotonic()
        self._pull(item.source_path, part)
        pull_s = time.monotonic() - t_pull
        n_bytes = part.stat().st_size

        # Hash once, compare once — the failure message reuses the digest, so
        # ``_verifies`` (which would re-hash) is deliberately not used here.
        verify_note = ""
        if item.sha256 is not None:
            t_hash = time.monotonic()
            actual = sha256_of(part)
            verify_note = f" + {time.monotonic() - t_hash:.2f}s verify"
            if actual != item.sha256:
                part.unlink()
                raise ChecksumError(item.source_path, item.sha256, actual)
        part.replace(dest)
        self._maybe_unarchive(item, dest)
        logger.debug(f"Fetched {item.rel_path}: {n_bytes} bytes in {pull_s:.2f}s transfer{verify_note}")
        return ("fetched", n_bytes)

    def _iter_attempts(
        self, items: list[RemoteFile], dest_dir: Path, do_overwrite: bool
    ) -> Iterator[tuple[RemoteFile, Callable[[], tuple[str, int]]]]:
        """Pair each manifest row with the zero-arg thunk that fetches it.

        The thunks close over everything :meth:`_fetch_one` needs, so the
        dispatch layer (:meth:`_attempts`) can treat sequential and pooled
        execution identically — it invokes (or submits) opaque callables and
        never needs the fetch arguments itself.
        """
        for item in items:
            yield item, partial(self._fetch_one, item, dest_dir, do_overwrite)

    @staticmethod
    def _attempts(
        items_to_fetch: Iterable[tuple[RemoteFile, Callable[[], tuple[str, int]]]],
        pool: Executor | None,
        abort: threading.Event,
    ) -> Iterator[tuple[RemoteFile, Callable[[], tuple[str, int]]]]:
        """Dispatch ``(item, callable)`` pairs sequentially or through a pool.

        Sequential mode yields the input pairs unchanged; the caller invokes
        them in the main thread. Parallel mode submits callables to ``pool``
        through a bounded sliding window — an initial batch of at most
        :data:`_MAX_PENDING_SUBMITS`, then one fresh submission per completion
        — and yields ``(item, future.result)`` pairs in completion order.
        Bounding the window keeps dispatch bookkeeping O(window) rather than
        O(manifest): each pending Future costs ~2 KB, which adds up to
        hundreds of MB on 100k+-row manifests if submitted all at once.

        Fail-fast in parallel mode: if the caller raises out of its loop on the first
        failure, the ``finally`` halts the rest of the bundle. That takes TWO
        mechanisms, because either alone leaks work:

        - ``Future.cancel()`` stops futures that have not started. It is a no-op on a
          future already running — there is no way to interrupt a thread mid-fetch.
        - An **abort flag**, checked at the top of every queued fetch, stops the ones
          that start between the failure surfacing and the cancel landing. Without it,
          a worker freed by the failure immediately picks up the next queued item and
          runs it to completion, and with a window of :data:`_MAX_PENDING_SUBMITS`
          there can be many such items — each one a real transfer in production.

        WHEN the flag is set is what makes it effective. ``download_all`` trips it the
        instant a fetch raises, before re-raising — not here in the ``finally``. Waiting
        for teardown leaves the worker free to drain queued items during the unwind:
        measured on a loaded box with microsecond-fast fetches, that leaked up to 15 of
        19 queued items. Tripping it at detection cuts the leak to the fetches already
        in flight, which nothing short of killing threads could stop.

        This ``finally`` still sets it, covering early exits that are not failures (a
        caller ``break``), and cancels the queue either way.

        The caller must wrap the generator in :func:`contextlib.closing` to guarantee
        the ``finally`` runs.
        """
        if pool is None:
            yield from items_to_fetch
            return
        item_iter = iter(items_to_fetch)

        def guarded(run: Callable[[], tuple[str, int]]) -> Callable[[], tuple[str, int]]:
            """Wrap a fetch so it becomes a no-op once fail-fast has tripped."""

            def _run() -> tuple[str, int]:
                if abort.is_set():
                    raise _AbortedError
                return run()

            return _run

        pending = {
            pool.submit(guarded(run)): item for item, run in itertools.islice(item_iter, _MAX_PENDING_SUBMITS)
        }
        try:
            while pending:
                done, _ = wait(pending, return_when=FIRST_COMPLETED)
                for fut in done:
                    item = pending.pop(fut)
                    # Replenish before yielding: if the caller raises out of this
                    # yield, the fresh submission is still queued and the
                    # ``finally`` below cancels it — same fail-fast semantics as
                    # the queued remainder of an up-front submission.
                    nxt = next(item_iter, None)
                    if nxt is not None:
                        nxt_item, nxt_run = nxt
                        pending[pool.submit(guarded(nxt_run))] = nxt_item
                    yield item, fut.result
        finally:
            # Flag first, then cancel: a future that slips past ``cancel`` still sees
            # the flag and returns without fetching.
            abort.set()
            for fut in pending:
                fut.cancel()

files cached property

The validated manifest — calls :meth:_list_files once, materializes, filters, and validates.

Cached on first access. Subsequent download_all calls reuse the same list rather than re-hitting :meth:_list_files (which may do network I/O — e.g. PhysioNet fetches SHA256SUMS.txt). If a source’s contents could change between runs and the caller wants a fresh manifest, build a new Source instance.

Per-row validation (relative, no traversal, no backslashes, well-formed sha256) happens at :class:RemoteFile construction inside :meth:_list_files, so it needs no re-checking here. This property adds the two whole-manifest steps:

  • include / exclude filtering — the constructor’s glob patterns are matched against each row’s normalized rel_path; rows an include list doesn’t match, or an exclude list does match, are dropped.
  • duplicate-destination detection — two rows whose normalized rel_paths collide (a/../x.csv vs x.csv) would race on the same .part file under concurrent workers, so they fail the bundle up-front.

_fetch_one calls :meth:_resolve_dest per-item at fetch time as the runtime security boundary (e.g. for dest_dirs that contain symlinks).

Examples:

Duplicate destinations are caught even when the strings differ:

>>> class DupSource(Source):
...     def _list_files(self):
...         return [RemoteFile("a.txt", ""), RemoteFile("sub/../a.txt", "")]
...     def _pull(self, source_path, target):
...         target.write_text("never reached")
>>>
>>> DupSource().files
Traceback (most recent call last):
    ...
ValueError: Duplicate destination 'sub/../a.txt': collides with 'a.txt'. ...

Unsafe rel_paths fail earlier still — at :class:RemoteFile construction inside _list_files:

>>> class EscapingSource(Source):
...     def _list_files(self):
...         return [RemoteFile("../escape.txt", "")]
...     def _pull(self, source_path, target):
...         target.write_text("never reached")
>>>
>>> EscapingSource().files
Traceback (most recent call last):
    ...
ValueError: rel_path '../escape.txt' escapes dest_dir ...

include / exclude globs subset the manifest:

>>> class TreeSource(Source):
...     def _list_files(self):
...         return [
...             RemoteFile("hosp/patients.csv.gz", ""),
...             RemoteFile("hosp/labevents.csv.gz", ""),
...             RemoteFile("note/discharge.csv.gz", ""),
...         ]
...     def _pull(self, source_path, target):
...         target.write_text("ok")
>>>
>>> [f.rel_path for f in TreeSource(include=["hosp/*"]).files]
['hosp/patients.csv.gz', 'hosp/labevents.csv.gz']
>>> [f.rel_path for f in TreeSource(exclude=["*/labevents*"]).files]
['hosp/patients.csv.gz', 'note/discharge.csv.gz']

exclude is applied after include — both together intersect:

>>> [f.rel_path for f in TreeSource(include=["hosp/*"], exclude=["*/labevents*"]).files]
['hosp/patients.csv.gz']

An empty include list matches nothing (fnmatch semantics) — it selects zero files rather than silently selecting everything:

>>> TreeSource(include=[]).files
[]

n_files property

Number of files in the validated, filtered manifest.

_attempts(items_to_fetch, pool, abort) staticmethod

Dispatch (item, callable) pairs sequentially or through a pool.

Sequential mode yields the input pairs unchanged; the caller invokes them in the main thread. Parallel mode submits callables to pool through a bounded sliding window — an initial batch of at most :data:_MAX_PENDING_SUBMITS, then one fresh submission per completion — and yields (item, future.result) pairs in completion order. Bounding the window keeps dispatch bookkeeping O(window) rather than O(manifest): each pending Future costs ~2 KB, which adds up to hundreds of MB on 100k+-row manifests if submitted all at once.

Fail-fast in parallel mode: if the caller raises out of its loop on the first failure, the finally halts the rest of the bundle. That takes TWO mechanisms, because either alone leaks work:

  • Future.cancel() stops futures that have not started. It is a no-op on a future already running — there is no way to interrupt a thread mid-fetch.
  • An abort flag, checked at the top of every queued fetch, stops the ones that start between the failure surfacing and the cancel landing. Without it, a worker freed by the failure immediately picks up the next queued item and runs it to completion, and with a window of :data:_MAX_PENDING_SUBMITS there can be many such items — each one a real transfer in production.

WHEN the flag is set is what makes it effective. download_all trips it the instant a fetch raises, before re-raising — not here in the finally. Waiting for teardown leaves the worker free to drain queued items during the unwind: measured on a loaded box with microsecond-fast fetches, that leaked up to 15 of 19 queued items. Tripping it at detection cuts the leak to the fetches already in flight, which nothing short of killing threads could stop.

This finally still sets it, covering early exits that are not failures (a caller break), and cancels the queue either way.

The caller must wrap the generator in :func:contextlib.closing to guarantee the finally runs.

Source code in MEDS_extract/download/source.py 
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
@staticmethod
def _attempts(
    items_to_fetch: Iterable[tuple[RemoteFile, Callable[[], tuple[str, int]]]],
    pool: Executor | None,
    abort: threading.Event,
) -> Iterator[tuple[RemoteFile, Callable[[], tuple[str, int]]]]:
    """Dispatch ``(item, callable)`` pairs sequentially or through a pool.

    Sequential mode yields the input pairs unchanged; the caller invokes
    them in the main thread. Parallel mode submits callables to ``pool``
    through a bounded sliding window — an initial batch of at most
    :data:`_MAX_PENDING_SUBMITS`, then one fresh submission per completion
    — and yields ``(item, future.result)`` pairs in completion order.
    Bounding the window keeps dispatch bookkeeping O(window) rather than
    O(manifest): each pending Future costs ~2 KB, which adds up to
    hundreds of MB on 100k+-row manifests if submitted all at once.

    Fail-fast in parallel mode: if the caller raises out of its loop on the first
    failure, the ``finally`` halts the rest of the bundle. That takes TWO
    mechanisms, because either alone leaks work:

    - ``Future.cancel()`` stops futures that have not started. It is a no-op on a
      future already running — there is no way to interrupt a thread mid-fetch.
    - An **abort flag**, checked at the top of every queued fetch, stops the ones
      that start between the failure surfacing and the cancel landing. Without it,
      a worker freed by the failure immediately picks up the next queued item and
      runs it to completion, and with a window of :data:`_MAX_PENDING_SUBMITS`
      there can be many such items — each one a real transfer in production.

    WHEN the flag is set is what makes it effective. ``download_all`` trips it the
    instant a fetch raises, before re-raising — not here in the ``finally``. Waiting
    for teardown leaves the worker free to drain queued items during the unwind:
    measured on a loaded box with microsecond-fast fetches, that leaked up to 15 of
    19 queued items. Tripping it at detection cuts the leak to the fetches already
    in flight, which nothing short of killing threads could stop.

    This ``finally`` still sets it, covering early exits that are not failures (a
    caller ``break``), and cancels the queue either way.

    The caller must wrap the generator in :func:`contextlib.closing` to guarantee
    the ``finally`` runs.
    """
    if pool is None:
        yield from items_to_fetch
        return
    item_iter = iter(items_to_fetch)

    def guarded(run: Callable[[], tuple[str, int]]) -> Callable[[], tuple[str, int]]:
        """Wrap a fetch so it becomes a no-op once fail-fast has tripped."""

        def _run() -> tuple[str, int]:
            if abort.is_set():
                raise _AbortedError
            return run()

        return _run

    pending = {
        pool.submit(guarded(run)): item for item, run in itertools.islice(item_iter, _MAX_PENDING_SUBMITS)
    }
    try:
        while pending:
            done, _ = wait(pending, return_when=FIRST_COMPLETED)
            for fut in done:
                item = pending.pop(fut)
                # Replenish before yielding: if the caller raises out of this
                # yield, the fresh submission is still queued and the
                # ``finally`` below cancels it — same fail-fast semantics as
                # the queued remainder of an up-front submission.
                nxt = next(item_iter, None)
                if nxt is not None:
                    nxt_item, nxt_run = nxt
                    pending[pool.submit(guarded(nxt_run))] = nxt_item
                yield item, fut.result
    finally:
        # Flag first, then cancel: a future that slips past ``cancel`` still sees
        # the flag and returns without fetching.
        abort.set()
        for fut in pending:
            fut.cancel()

_fetch_one(item, dest_dir, do_overwrite)

Fetch one manifest entry end-to-end: policy → .part staging → verify → rename.

Pipeline:

  1. Resolve dest = dest_dir / item.rel_path (with traversal validation).

  2. If do_overwrite=True: unconditionally clear dest and any stale .part (whether or not dest exists), then proceed to step 5. Otherwise, on a pre-existing dest:

  3. dest verifies against item.sha256: skip and return.

  4. item.sha256 is set but dest mismatches: re-fetch (with a warning naming the file) — dest is replaced only by the atomic rename in step 7, after the fresh bytes verify.

  5. no item.sha256: re-fetch — a file we can’t prove matches the manifest is never trusted, and never skipped.

  6. If a prior run left a .part that already verifies against item.sha256 (interrupted between the last byte and the rename), promote it to dest directly — no re-fetch.

  7. If the manifest has no SHA to verify against, discard any stale .part — resume-without-verification is unsafe.
  8. Call self._pull(item.source_path, part) — backend streams bytes.
  9. If item.sha256 is set, hash part once via :func:sha256_of and compare; on mismatch, unlink part and raise :class:ChecksumError.
  10. Atomic-rename part → dest.
  11. If item.unarchive is set, run the post-fetch unpack hook (:meth:_maybe_unarchive) — also applied on the promote path in step 3, but never to a step-2 skip (the dest was not newly written).

Returns:

Type Description
str

A (status, n_bytes) tuple where status is "skipped"
int

(already-complete dest), "promoted" (complete .part
tuple[str, int]

renamed without re-fetching), or "fetched" (bytes actually
tuple[str, int]

transferred), and n_bytes is the size of the file the status
tuple[str, int]

applies to. download_all tallies these into its end-of-bundle
tuple[str, int]

summary.

On any exception, no new dest is created and an existing dest is never modified — the re-fetch paths replace it only via the atomic rename, after the fresh bytes verify. part may exist after a partial transport failure (intentional — gives a future run a head start via Range-resume on backends that support it).

Examples:

Backend’s _pull produces the bytes; this method handles staging, sha verification, and atomic rename:

>>> import hashlib
>>> class FakeSource(Source):
...     def _list_files(self):
...         return [RemoteFile("x.txt", "dummy", sha256=hashlib.sha256(b"hi").hexdigest())]
...     def _pull(self, source_path, target):
...         target.write_bytes(b"hi")
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     src = FakeSource()
...     [item] = src.files
...     src._fetch_one(item, d, do_overwrite=False)
...     print((d / "x.txt").read_bytes(), (d / "x.txt.part").exists())
('fetched', 2)
b'hi' False

On a SHA mismatch the staged .part is deleted, dest is not created, and :class:ChecksumError propagates:

>>> class WrongShaSource(Source):
...     def _list_files(self):
...         return [RemoteFile("x.txt", "dummy", sha256="0" * 64)]
...     def _pull(self, source_path, target):
...         target.write_bytes(b"hi")
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     src = WrongShaSource()
...     [item] = src.files
...     try:
...         src._fetch_one(item, d, do_overwrite=False)
...     except ChecksumError:
...         print(f"raised; dest={(d / 'x.txt').exists()}, part={(d / 'x.txt.part').exists()}")
raised; dest=False, part=False

When the manifest has no SHA, a stale .part from a prior failed run can’t be safely resumed (nothing would catch silent corruption), so it is discarded before _pull runs — the backend starts fresh:

>>> class NoShaSource(Source):
...     def _list_files(self):
...         return [RemoteFile("x.txt", "dummy")]  # no sha
...     def _pull(self, source_path, target):
...         print(f"stale .part visible to _pull: {target.exists()}")
...         target.write_text("fresh")
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     _ = (d / "x.txt.part").write_bytes(b"stale partial")
...     src = NoShaSource()
...     [item] = src.files
...     src._fetch_one(item, d, do_overwrite=False)
stale .part visible to _pull: False
('fetched', 5)

A leftover .part that already verifies against the manifest sha (prior run died between the last byte and the rename) is promoted to dest directly — _pull is never invoked:

>>> body = b"the whole file, fully written"
>>> class NoRefetchSource(Source):
...     def _list_files(self):
...         return [RemoteFile("x.txt", "dummy", sha256=hashlib.sha256(body).hexdigest())]
...     def _pull(self, source_path, target):
...         raise AssertionError("must not re-fetch a complete .part")
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     _ = (d / "x.txt.part").write_bytes(body)
...     src = NoRefetchSource()
...     [item] = src.files
...     src._fetch_one(item, d, do_overwrite=False)
...     print((d / "x.txt").read_bytes() == body, (d / "x.txt.part").exists())
('promoted', 29)
True False
Source code in MEDS_extract/download/source.py 
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
def _fetch_one(self, item: RemoteFile, dest_dir: Path, do_overwrite: bool) -> tuple[str, int]:
    """Fetch one manifest entry end-to-end: policy → ``.part`` staging → verify → rename.

    Pipeline:

    1. Resolve ``dest = dest_dir / item.rel_path`` (with traversal validation).
    2. If ``do_overwrite=True``: unconditionally clear ``dest`` and any stale
       ``.part`` (whether or not ``dest`` exists), then proceed to step 5.
       Otherwise, on a pre-existing ``dest``:

       - ``dest`` verifies against ``item.sha256``: skip and return.
       - ``item.sha256`` is set but ``dest`` mismatches: re-fetch (with a
         warning naming the file) — ``dest`` is replaced only by the atomic
         rename in step 7, after the fresh bytes verify.
       - no ``item.sha256``: re-fetch — a file we can't prove matches the
         manifest is never trusted, and never skipped.

    3. If a prior run left a ``.part`` that already verifies against
       ``item.sha256`` (interrupted between the last byte and the rename),
       promote it to ``dest`` directly — no re-fetch.
    4. If the manifest has no SHA to verify against, discard any stale
       ``.part`` — resume-without-verification is unsafe.
    5. Call ``self._pull(item.source_path, part)`` — backend streams bytes.
    6. If ``item.sha256`` is set, hash ``part`` once via :func:`sha256_of`
       and compare; on mismatch, unlink ``part`` and raise
       :class:`ChecksumError`.
    7. Atomic-rename ``part`` → ``dest``.
    8. If ``item.unarchive`` is set, run the post-fetch unpack hook
       (:meth:`_maybe_unarchive`) — also applied on the promote path in
       step 3, but never to a step-2 skip (the dest was not newly written).

    Returns:
        A ``(status, n_bytes)`` tuple where ``status`` is ``"skipped"``
        (already-complete dest), ``"promoted"`` (complete ``.part``
        renamed without re-fetching), or ``"fetched"`` (bytes actually
        transferred), and ``n_bytes`` is the size of the file the status
        applies to. ``download_all`` tallies these into its end-of-bundle
        summary.

    On any exception, no new ``dest`` is created and an existing ``dest``
    is never modified — the re-fetch paths replace it only via the atomic
    rename, after the fresh bytes verify. ``part`` may exist after a
    partial transport failure (intentional — gives a future run a head
    start via Range-resume on backends that support it).

    Examples:
        Backend's ``_pull`` produces the bytes; this method handles staging,
        sha verification, and atomic rename:

        >>> import hashlib
        >>> class FakeSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.txt", "dummy", sha256=hashlib.sha256(b"hi").hexdigest())]
        ...     def _pull(self, source_path, target):
        ...         target.write_bytes(b"hi")
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     src = FakeSource()
        ...     [item] = src.files
        ...     src._fetch_one(item, d, do_overwrite=False)
        ...     print((d / "x.txt").read_bytes(), (d / "x.txt.part").exists())
        ('fetched', 2)
        b'hi' False

        On a SHA mismatch the staged ``.part`` is deleted, ``dest`` is not
        created, and :class:`ChecksumError` propagates:

        >>> class WrongShaSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.txt", "dummy", sha256="0" * 64)]
        ...     def _pull(self, source_path, target):
        ...         target.write_bytes(b"hi")
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     src = WrongShaSource()
        ...     [item] = src.files
        ...     try:
        ...         src._fetch_one(item, d, do_overwrite=False)
        ...     except ChecksumError:
        ...         print(f"raised; dest={(d / 'x.txt').exists()}, part={(d / 'x.txt.part').exists()}")
        raised; dest=False, part=False

        When the manifest has no SHA, a stale ``.part`` from a prior failed
        run can't be safely resumed (nothing would catch silent corruption),
        so it is discarded before ``_pull`` runs — the backend starts fresh:

        >>> class NoShaSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.txt", "dummy")]  # no sha
        ...     def _pull(self, source_path, target):
        ...         print(f"stale .part visible to _pull: {target.exists()}")
        ...         target.write_text("fresh")
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     _ = (d / "x.txt.part").write_bytes(b"stale partial")
        ...     src = NoShaSource()
        ...     [item] = src.files
        ...     src._fetch_one(item, d, do_overwrite=False)
        stale .part visible to _pull: False
        ('fetched', 5)

        A leftover ``.part`` that already verifies against the manifest sha
        (prior run died between the last byte and the rename) is promoted to
        ``dest`` directly — ``_pull`` is never invoked:

        >>> body = b"the whole file, fully written"
        >>> class NoRefetchSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.txt", "dummy", sha256=hashlib.sha256(body).hexdigest())]
        ...     def _pull(self, source_path, target):
        ...         raise AssertionError("must not re-fetch a complete .part")
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     _ = (d / "x.txt.part").write_bytes(body)
        ...     src = NoRefetchSource()
        ...     [item] = src.files
        ...     src._fetch_one(item, d, do_overwrite=False)
        ...     print((d / "x.txt").read_bytes() == body, (d / "x.txt.part").exists())
        ('promoted', 29)
        True False
    """
    dest = self._resolve_dest(dest_dir, item.rel_path)
    dest.parent.mkdir(parents=True, exist_ok=True)
    part = dest.with_name(dest.name + ".part")

    if do_overwrite:
        # Clear both independently — a ``.part`` from a half-finished prior
        # run can exist even when ``dest`` doesn't, and either one left in
        # place would be picked up as a Range-resume base.
        if dest.exists():
            dest.unlink()
        if part.exists():
            part.unlink()
    elif dest.exists():
        if self._verifies(dest, item):
            logger.debug(f"Skipping {item.rel_path}: already complete.")
            return ("skipped", dest.stat().st_size)
        # An existing dest that can't be proven complete is never trusted and
        # never skipped: re-fetch it. The stale copy stays in place until the
        # fresh bytes verify — the atomic rename below is what replaces it.
        if item.sha256 is None:
            logger.debug(f"Re-fetching {item.rel_path}: existing file has no manifest sha to verify.")
        else:
            logger.warning(
                f"Re-fetching {item.rel_path}: existing file at {dest} failed SHA-256 "
                "verification against the manifest."
            )

    if item.sha256 is not None:
        # A prior run may have died between writing the last byte and the
        # rename below — in that case the ``.part`` is the complete file and
        # re-fetching it (or bouncing off an unsatisfiable Range request)
        # wastes the whole transfer. Verify and promote directly.
        if part.exists() and sha256_of(part) == item.sha256:
            logger.debug(f"Promoting complete .part for {item.rel_path} without re-fetching.")
            n_bytes = part.stat().st_size
            part.replace(dest)
            self._maybe_unarchive(item, dest)
            return ("promoted", n_bytes)
    elif part.exists():
        # Resume-without-verification is unsafe: without a sha to catch silent
        # corruption, a stale ``.part`` could be from a different version of
        # the source file. Clear it so ``_pull`` starts fresh. With sha set,
        # Range-resume is safe because the post-write verify catches mismatches.
        part.unlink()

    t_pull = time.monotonic()
    self._pull(item.source_path, part)
    pull_s = time.monotonic() - t_pull
    n_bytes = part.stat().st_size

    # Hash once, compare once — the failure message reuses the digest, so
    # ``_verifies`` (which would re-hash) is deliberately not used here.
    verify_note = ""
    if item.sha256 is not None:
        t_hash = time.monotonic()
        actual = sha256_of(part)
        verify_note = f" + {time.monotonic() - t_hash:.2f}s verify"
        if actual != item.sha256:
            part.unlink()
            raise ChecksumError(item.source_path, item.sha256, actual)
    part.replace(dest)
    self._maybe_unarchive(item, dest)
    logger.debug(f"Fetched {item.rel_path}: {n_bytes} bytes in {pull_s:.2f}s transfer{verify_note}")
    return ("fetched", n_bytes)

_iter_attempts(items, dest_dir, do_overwrite)

Pair each manifest row with the zero-arg thunk that fetches it.

The thunks close over everything :meth:_fetch_one needs, so the dispatch layer (:meth:_attempts) can treat sequential and pooled execution identically — it invokes (or submits) opaque callables and never needs the fetch arguments itself.

Source code in MEDS_extract/download/source.py 
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
def _iter_attempts(
    self, items: list[RemoteFile], dest_dir: Path, do_overwrite: bool
) -> Iterator[tuple[RemoteFile, Callable[[], tuple[str, int]]]]:
    """Pair each manifest row with the zero-arg thunk that fetches it.

    The thunks close over everything :meth:`_fetch_one` needs, so the
    dispatch layer (:meth:`_attempts`) can treat sequential and pooled
    execution identically — it invokes (or submits) opaque callables and
    never needs the fetch arguments itself.
    """
    for item in items:
        yield item, partial(self._fetch_one, item, dest_dir, do_overwrite)

_list_files() abstractmethod

Subclass hook — enumerate the files this source offers.

:attr:files is the validating cached wrapper that callers use; this hook just produces the rows.

Source code in MEDS_extract/download/source.py 
664
665
666
667
668
669
670
@abstractmethod
def _list_files(self) -> Iterable[RemoteFile]:
    """Subclass hook — enumerate the files this source offers.

    :attr:`files` is the validating cached wrapper that callers use; this
    hook just produces the rows.
    """

_maybe_unarchive(item, dest) staticmethod

Post-fetch unpack hook — runs after bytes newly land at dest.

A no-op unless item.unarchive is set. "auto" resolves the format from dest’s extension (and is a no-op on non-archive extensions like .csv.gz); explicit formats dispatch directly. Extraction lands in dest’s directory via :func:~MEDS_extract.download.unarchive.safe_extract, which validates every member against zip-slip / tar-slip before any bytes are written.

Tri-state cleanup: cleanup_archive=None defers to the unarchive mode — :attr:~MEDS_extract.download.unarchive.ArchiveFormat.AUTO removes the archive (the one-arg “fetch + extract + drop” flow), explicit formats keep it. Explicit True / False always wins.

Invoked from :meth:_fetch_one on the "fetched" and "promoted" paths only — a "skipped" dest was not newly written, so it is not re-extracted.

Examples:

unarchive="auto" unpacks a zip next to itself and (by AUTO’s cleanup default) removes the archive afterwards:

>>> import zipfile
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     archive = d / "bundle.zip"
...     with zipfile.ZipFile(archive, "w") as zf:
...         zf.writestr("sub/a.csv", "col\n1")
...     Source._maybe_unarchive(RemoteFile("bundle.zip", "", unarchive="auto"), archive)
...     print_directory(d)
└── sub
    └── a.csv

An explicit format keeps the archive by default:

>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     archive = d / "bundle.zip"
...     with zipfile.ZipFile(archive, "w") as zf:
...         zf.writestr("a.csv", "col\n1")
...     Source._maybe_unarchive(RemoteFile("bundle.zip", "", unarchive="zip"), archive)
...     print_directory(d)
├── a.csv
└── bundle.zip

cleanup_archive overrides the mode default in either direction:

>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     archive = d / "bundle.zip"
...     with zipfile.ZipFile(archive, "w") as zf:
...         zf.writestr("a.csv", "col\n1")
...     item = RemoteFile("bundle.zip", "", unarchive="zip", cleanup_archive=True)
...     Source._maybe_unarchive(item, archive)
...     print_directory(d)
└── a.csv

"auto" on a non-archive is a no-op — the file is left alone:

>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     f = d / "patients.csv.gz"
...     _ = f.write_bytes(b"not an archive")
...     Source._maybe_unarchive(RemoteFile("patients.csv.gz", "", unarchive="auto"), f)
...     print_directory(d)
└── patients.csv.gz
Source code in MEDS_extract/download/source.py 
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
@staticmethod
def _maybe_unarchive(item: RemoteFile, dest: Path) -> None:
    """Post-fetch unpack hook — runs after bytes newly land at ``dest``.

    A no-op unless ``item.unarchive`` is set. ``"auto"`` resolves the format
    from ``dest``'s extension (and is a no-op on non-archive extensions like
    ``.csv.gz``); explicit formats dispatch directly. Extraction lands in
    ``dest``'s directory via
    :func:`~MEDS_extract.download.unarchive.safe_extract`, which validates
    every member against zip-slip / tar-slip before any bytes are written.

    Tri-state cleanup: ``cleanup_archive=None`` defers to the unarchive mode —
    :attr:`~MEDS_extract.download.unarchive.ArchiveFormat.AUTO` removes the
    archive (the one-arg "fetch + extract + drop" flow), explicit formats keep
    it. Explicit ``True`` / ``False`` always wins.

    Invoked from :meth:`_fetch_one` on the ``"fetched"`` and ``"promoted"``
    paths only — a ``"skipped"`` dest was not newly written, so it is not
    re-extracted.

    Examples:
        ``unarchive="auto"`` unpacks a zip next to itself and (by AUTO's
        cleanup default) removes the archive afterwards:

        >>> import zipfile
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     archive = d / "bundle.zip"
        ...     with zipfile.ZipFile(archive, "w") as zf:
        ...         zf.writestr("sub/a.csv", "col\\n1")
        ...     Source._maybe_unarchive(RemoteFile("bundle.zip", "", unarchive="auto"), archive)
        ...     print_directory(d)
        └── sub
            └── a.csv

        An explicit format keeps the archive by default:

        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     archive = d / "bundle.zip"
        ...     with zipfile.ZipFile(archive, "w") as zf:
        ...         zf.writestr("a.csv", "col\\n1")
        ...     Source._maybe_unarchive(RemoteFile("bundle.zip", "", unarchive="zip"), archive)
        ...     print_directory(d)
        ├── a.csv
        └── bundle.zip

        ``cleanup_archive`` overrides the mode default in either direction:

        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     archive = d / "bundle.zip"
        ...     with zipfile.ZipFile(archive, "w") as zf:
        ...         zf.writestr("a.csv", "col\\n1")
        ...     item = RemoteFile("bundle.zip", "", unarchive="zip", cleanup_archive=True)
        ...     Source._maybe_unarchive(item, archive)
        ...     print_directory(d)
        └── a.csv

        ``"auto"`` on a non-archive is a no-op — the file is left alone:

        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     f = d / "patients.csv.gz"
        ...     _ = f.write_bytes(b"not an archive")
        ...     Source._maybe_unarchive(RemoteFile("patients.csv.gz", "", unarchive="auto"), f)
        ...     print_directory(d)
        └── patients.csv.gz
    """
    if not item.unarchive:
        return
    fmt = resolve_format(item.unarchive, dest)
    if fmt is None:
        return
    t0 = time.monotonic()
    safe_extract(dest, dest.parent, fmt)
    logger.debug(f"Extracted {item.rel_path} ({fmt.value}) in {time.monotonic() - t0:.1f}s")
    if item.cleanup_archive is None:
        cleanup = ArchiveFormat(item.unarchive) is ArchiveFormat.AUTO
    else:
        cleanup = item.cleanup_archive
    if cleanup:
        dest.unlink()

_pull(source_path, target) abstractmethod

Stream the bytes at source_path into target.

source_path is whatever the backend stored in RemoteFile.source_path when it built the manifest (a URL for HTTP, a UPath spec for fsspec). On successful return target contains the complete file; on any transport error, raise.

Backends with resume semantics (HTTP Range) MAY observe existing bytes at target and append; backends without resume should overwrite.

Source code in MEDS_extract/download/source.py 
672
673
674
675
676
677
678
679
680
681
682
683
684
@abstractmethod
def _pull(self, source_path: str, target: Path) -> None:
    """Stream the bytes at ``source_path`` into ``target``.

    ``source_path`` is whatever the backend stored in
    ``RemoteFile.source_path`` when it built the manifest (a URL for HTTP,
    a UPath spec for fsspec). On successful return ``target`` contains
    the complete file; on any transport error, raise.

    Backends with resume semantics (HTTP ``Range``) MAY observe existing
    bytes at ``target`` and append; backends without resume should
    overwrite.
    """

_resolve_dest(dest_dir, rel_path) staticmethod

Resolve rel_path under dest_dir, rejecting any escape attempts.

:class:RemoteFile construction already rejects malformed rel_paths by string inspection; this fetch-time check is the runtime security boundary against escapes that only materialize on a real filesystem (e.g. symlinks inside dest_dir).

Source code in MEDS_extract/download/source.py 
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
@staticmethod
def _resolve_dest(dest_dir: Path, rel_path: str) -> Path:
    """Resolve ``rel_path`` under ``dest_dir``, rejecting any escape attempts.

    :class:`RemoteFile` construction already rejects malformed rel_paths by
    string inspection; this fetch-time check is the runtime security boundary
    against escapes that only materialize on a real filesystem (e.g. symlinks
    inside ``dest_dir``).
    """
    if Path(rel_path).is_absolute():
        raise ValueError(f"rel_path must be relative, got absolute: {rel_path!r}")
    resolved, contained = resolve_contained(dest_dir, rel_path)
    if not contained:
        raise ValueError(
            f"rel_path {rel_path!r} escapes dest_dir {Path(dest_dir).resolve()} (resolved to {resolved})."
        )
    return resolved

_selected(item)

Apply the constructor’s include / exclude globs to one manifest row.

Source code in MEDS_extract/download/source.py 
651
652
653
def _selected(self, item: RemoteFile) -> bool:
    """Apply the constructor's ``include`` / ``exclude`` globs to one manifest row."""
    return self._selected_path(item.dest_key)

_selected_path(dest_key)

String-level filter check, for backends that want to skip expensive per-file work (e.g. hashing) on rows the manifest filters would drop anyway.

Source code in MEDS_extract/download/source.py 
655
656
657
658
659
660
661
662
def _selected_path(self, dest_key: str) -> bool:
    """String-level filter check, for backends that want to skip expensive per-file work (e.g. hashing) on
    rows the manifest filters would drop anyway."""
    if self._include is not None and not any(fnmatch.fnmatchcase(dest_key, p) for p in self._include):
        return False
    return not (
        self._exclude is not None and any(fnmatch.fnmatchcase(dest_key, p) for p in self._exclude)
    )

_verifies(dest, item) staticmethod

True iff dest exists AND the manifest’s sha256 matches.

SHA-256 is the only verifier we trust. Same-size files can have different content; existence-with-no-hash means the file on disk could be anything. Backends that want skip-on-rerun semantics must populate sha256.

Source code in MEDS_extract/download/source.py 
717
718
719
720
721
722
723
724
725
@staticmethod
def _verifies(dest: Path, item: RemoteFile) -> bool:
    """True iff ``dest`` exists AND the manifest's ``sha256`` matches.

    SHA-256 is the only verifier we trust. Same-size files can have different
    content; existence-with-no-hash means the file on disk could be anything.
    Backends that want skip-on-rerun semantics must populate ``sha256``.
    """
    return item.sha256 is not None and dest.exists() and sha256_of(dest) == item.sha256

close()

Release transport resources held by this source.

Default is a no-op. Subclasses that own network clients / file handles / connection pools override this. Safe to call multiple times; safe to call on sources that own nothing.

Source code in MEDS_extract/download/source.py 
686
687
688
689
690
691
def close(self) -> None:  # noqa: B027 — intentional no-op default; subclasses override when needed
    """Release transport resources held by this source.

    Default is a no-op. Subclasses that own network clients / file handles / connection pools override
    this. Safe to call multiple times; safe to call on sources that own nothing.
    """

download_all(dest_dir, *, pool=None, continue_on_error=False, do_overwrite=False)

Download every file this source lists into dest_dir.

Parameters:

Name Type Description Default
dest_dir str | Path

Where files land. Created if missing.

 required
pool Executor | None

Optional :class:~concurrent.futures.Executor (typically a :class:~concurrent.futures.ThreadPoolExecutor) to submit work to. The caller owns the pool’s lifetime. When None (default), the bundle is fetched sequentially in the calling thread — no thread pool is created. Pass a pool when you want parallelism, sized to whatever your transport tolerates.

 None
continue_on_error bool

If False (default), the first per-file failure propagates. If True, per-file errors are collected and raised as a single :class:ExceptionGroup at the end so the caller sees every failure, not just the first.

 False
do_overwrite bool

If True, skip the verified-dest check and clear dest / .part before each fetch — re-fetches everything from scratch, even files whose local copy verifies.

 False

Raises:

Type Description
Exception

From the transport layer on any per-file failure when continue_on_error=False.
ExceptionGroup

When continue_on_error=True and at least one file failed.
ValueError

When the manifest contains an unsafe rel_path (raised at :class:RemoteFile construction) or duplicate destinations (raised by :attr:files).

Examples:

Simple case — no pool passed, download_all runs sequentially:

>>> class StubSource(Source):
...     def _list_files(self):
...         return [RemoteFile("a.txt", ""), RemoteFile("sub/b.txt", "")]
...     def _pull(self, source_path, target):
...         target.write_text(f"contents of {target.name}")
>>>
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     StubSource().download_all(d)
...     print_directory(d)
├── a.txt
└── sub
    └── b.txt

Multi-source case — caller owns one :class:ThreadPoolExecutor and hands it to every source. Two sources writing distinct files into one dest_dir is the typical CLI pattern (one physionet source plus one http source for a metadata bundle):

>>> from concurrent.futures import ThreadPoolExecutor
>>>
>>> class SourceA(Source):
...     def _list_files(self):
...         return [RemoteFile("a.txt", "")]
...     def _pull(self, source_path, target):
...         target.write_text("from A")
>>>
>>> class SourceB(Source):
...     def _list_files(self):
...         return [RemoteFile("metadata/b.csv", "")]
...     def _pull(self, source_path, target):
...         target.write_text("from B")
>>>
>>> with tempfile.TemporaryDirectory() as d, ThreadPoolExecutor(max_workers=4) as pool:
...     d = Path(d)
...     for src in [SourceA(), SourceB()]:
...         src.download_all(d, pool=pool)
...     print_directory(d)
├── a.txt
└── metadata
    └── b.csv

Already-complete files are skipped — _pull is not invoked for any :class:RemoteFile whose on-disk copy verifies against the manifest’s sha256:

>>> import hashlib
>>> body = b"abc"
>>> digest = hashlib.sha256(body).hexdigest()
>>>
>>> class SkipSource(Source):
...     def _list_files(self):
...         return [RemoteFile("x.txt", "", sha256=digest)]
...     def _pull(self, source_path, target):
...         raise RuntimeError("must not be called — file is already complete")
>>>
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     _ = (d / "x.txt").write_bytes(body)
...     SkipSource().download_all(d)  # no exception → already-complete skip worked

do_overwrite=True re-fetches even a file whose on-disk copy verifies — _pull runs exactly once across the two calls below (skipped without overwrite, forced with it):

>>> pulls = []
>>> class CountingSource(Source):
...     def _list_files(self):
...         return [RemoteFile("x.txt", "", sha256=digest)]
...     def _pull(self, source_path, target):
...         pulls.append(source_path)
...         target.write_bytes(body)
>>>
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     _ = (d / "x.txt").write_bytes(body)
...     CountingSource().download_all(d)  # verified on disk → skipped
...     CountingSource().download_all(d, do_overwrite=True)  # forced re-fetch
...     len(pulls)
1

An existing dest with no manifest sha can never be proven complete, so it is re-fetched rather than trusted — the stale local copy is replaced atomically, and re-runs stay idempotent:

>>> class UnverifiableSource(Source):
...     def _list_files(self):
...         return [RemoteFile("x.txt", "")]  # no sha
...     def _pull(self, source_path, target):
...         target.write_text("fresh")
>>>
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     _ = (d / "x.txt").write_bytes(b"stale")
...     UnverifiableSource().download_all(d)
...     UnverifiableSource().download_all(d)  # re-run: re-fetches again, no error
...     print((d / "x.txt").read_text())
fresh

An existing dest whose content mismatches the manifest sha is likewise re-fetched (with a warning naming the file) — and the fresh bytes must still verify before the atomic replace:

>>> class MismatchRepairSource(Source):
...     def _list_files(self):
...         return [RemoteFile("x.txt", "", sha256=digest)]
...     def _pull(self, source_path, target):
...         target.write_bytes(body)
>>>
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     _ = (d / "x.txt").write_bytes(b"corrupt local copy")
...     MismatchRepairSource().download_all(d)
...     print((d / "x.txt").read_bytes() == body)
True

Failure policy: by default the first per-file failure propagates and later files are not attempted; continue_on_error=True attempts everything and collects the failures into one :class:ExceptionGroup:

>>> class FlakySource(Source):
...     def _list_files(self):
...         return [RemoteFile("bad.txt", "bad"), RemoteFile("good.txt", "ok")]
...     def _pull(self, source_path, target):
...         if source_path == "bad":
...             raise RuntimeError("transport boom")
...         target.write_text("ok")
>>>
>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     try:
...         FlakySource().download_all(d)
...     except RuntimeError as e:
...         print(f"raised: {e}; good.txt fetched: {(d / 'good.txt').exists()}")
raised: transport boom; good.txt fetched: False

>>> with tempfile.TemporaryDirectory() as d:
...     d = Path(d)
...     try:
...         FlakySource().download_all(d, continue_on_error=True)
...     except ExceptionGroup as eg:
...         print(f"{len(eg.exceptions)} failed; good.txt fetched: {(d / 'good.txt').exists()}")
1 failed; good.txt fetched: True

Path-traversal manifests and absolute paths are rejected at :class:RemoteFile construction; duplicate destinations are rejected at :attr:files (the first thing download_all accesses) — see those docstrings for examples.

Source code in MEDS_extract/download/source.py 
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
def download_all(
    self,
    dest_dir: str | Path,
    *,
    pool: Executor | None = None,
    continue_on_error: bool = False,
    do_overwrite: bool = False,
) -> None:
    """Download every file this source lists into ``dest_dir``.

    Args:
        dest_dir: Where files land. Created if missing.
        pool: Optional :class:`~concurrent.futures.Executor` (typically a
            :class:`~concurrent.futures.ThreadPoolExecutor`) to submit work
            to. The caller owns the pool's lifetime. When ``None`` (default),
            the bundle is fetched sequentially in the calling thread — no
            thread pool is created. Pass a pool when you want parallelism,
            sized to whatever your transport tolerates.
        continue_on_error: If ``False`` (default), the first per-file failure
            propagates. If ``True``, per-file errors are collected and raised
            as a single :class:`ExceptionGroup` at the end so the caller sees
            every failure, not just the first.
        do_overwrite: If ``True``, skip the verified-dest check and clear
            ``dest`` / ``.part`` before each fetch — re-fetches everything
            from scratch, even files whose local copy verifies.

    Raises:
        Exception: From the transport layer on any per-file failure when
            ``continue_on_error=False``.
        ExceptionGroup: When ``continue_on_error=True`` and at least one
            file failed.
        ValueError: When the manifest contains an unsafe rel_path (raised at
            :class:`RemoteFile` construction) or duplicate destinations
            (raised by :attr:`files`).

    Examples:
        Simple case — no pool passed, ``download_all`` runs sequentially:

        >>> class StubSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("a.txt", ""), RemoteFile("sub/b.txt", "")]
        ...     def _pull(self, source_path, target):
        ...         target.write_text(f"contents of {target.name}")
        >>>
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     StubSource().download_all(d)
        ...     print_directory(d)
        ├── a.txt
        └── sub
            └── b.txt

        Multi-source case — caller owns one :class:`ThreadPoolExecutor` and
        hands it to every source. Two sources writing distinct files into one
        ``dest_dir`` is the typical CLI pattern (one ``physionet`` source plus
        one ``http`` source for a metadata bundle):

        >>> from concurrent.futures import ThreadPoolExecutor
        >>>
        >>> class SourceA(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("a.txt", "")]
        ...     def _pull(self, source_path, target):
        ...         target.write_text("from A")
        >>>
        >>> class SourceB(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("metadata/b.csv", "")]
        ...     def _pull(self, source_path, target):
        ...         target.write_text("from B")
        >>>
        >>> with tempfile.TemporaryDirectory() as d, ThreadPoolExecutor(max_workers=4) as pool:
        ...     d = Path(d)
        ...     for src in [SourceA(), SourceB()]:
        ...         src.download_all(d, pool=pool)
        ...     print_directory(d)
        ├── a.txt
        └── metadata
            └── b.csv

        Already-complete files are skipped — ``_pull`` is not invoked for any
        :class:`RemoteFile` whose on-disk copy verifies against the manifest's
        ``sha256``:

        >>> import hashlib
        >>> body = b"abc"
        >>> digest = hashlib.sha256(body).hexdigest()
        >>>
        >>> class SkipSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.txt", "", sha256=digest)]
        ...     def _pull(self, source_path, target):
        ...         raise RuntimeError("must not be called — file is already complete")
        >>>
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     _ = (d / "x.txt").write_bytes(body)
        ...     SkipSource().download_all(d)  # no exception → already-complete skip worked

        ``do_overwrite=True`` re-fetches even a file whose on-disk copy
        verifies — ``_pull`` runs exactly once across the two calls below
        (skipped without overwrite, forced with it):

        >>> pulls = []
        >>> class CountingSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.txt", "", sha256=digest)]
        ...     def _pull(self, source_path, target):
        ...         pulls.append(source_path)
        ...         target.write_bytes(body)
        >>>
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     _ = (d / "x.txt").write_bytes(body)
        ...     CountingSource().download_all(d)  # verified on disk → skipped
        ...     CountingSource().download_all(d, do_overwrite=True)  # forced re-fetch
        ...     len(pulls)
        1

        An existing ``dest`` with **no manifest sha** can never be proven
        complete, so it is re-fetched rather than trusted — the stale local
        copy is replaced atomically, and re-runs stay idempotent:

        >>> class UnverifiableSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.txt", "")]  # no sha
        ...     def _pull(self, source_path, target):
        ...         target.write_text("fresh")
        >>>
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     _ = (d / "x.txt").write_bytes(b"stale")
        ...     UnverifiableSource().download_all(d)
        ...     UnverifiableSource().download_all(d)  # re-run: re-fetches again, no error
        ...     print((d / "x.txt").read_text())
        fresh

        An existing ``dest`` whose content **mismatches** the manifest sha is
        likewise re-fetched (with a warning naming the file) — and the fresh
        bytes must still verify before the atomic replace:

        >>> class MismatchRepairSource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.txt", "", sha256=digest)]
        ...     def _pull(self, source_path, target):
        ...         target.write_bytes(body)
        >>>
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     _ = (d / "x.txt").write_bytes(b"corrupt local copy")
        ...     MismatchRepairSource().download_all(d)
        ...     print((d / "x.txt").read_bytes() == body)
        True

        Failure policy: by default the first per-file failure propagates and
        later files are not attempted; ``continue_on_error=True`` attempts
        everything and collects the failures into one :class:`ExceptionGroup`:

        >>> class FlakySource(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("bad.txt", "bad"), RemoteFile("good.txt", "ok")]
        ...     def _pull(self, source_path, target):
        ...         if source_path == "bad":
        ...             raise RuntimeError("transport boom")
        ...         target.write_text("ok")
        >>>
        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     try:
        ...         FlakySource().download_all(d)
        ...     except RuntimeError as e:
        ...         print(f"raised: {e}; good.txt fetched: {(d / 'good.txt').exists()}")
        raised: transport boom; good.txt fetched: False

        >>> with tempfile.TemporaryDirectory() as d:
        ...     d = Path(d)
        ...     try:
        ...         FlakySource().download_all(d, continue_on_error=True)
        ...     except ExceptionGroup as eg:
        ...         print(f"{len(eg.exceptions)} failed; good.txt fetched: {(d / 'good.txt').exists()}")
        1 failed; good.txt fetched: True

        Path-traversal manifests and absolute paths are rejected at
        :class:`RemoteFile` construction; duplicate destinations are rejected
        at :attr:`files` (the first thing ``download_all`` accesses) — see
        those docstrings for examples.
    """
    # Materialize + validate the manifest before touching the filesystem, so a
    # malformed ``sources:`` entry doesn't leave behind an empty ``dest_dir``.
    items = self.files
    dest_dir = Path(dest_dir)
    dest_dir.mkdir(parents=True, exist_ok=True)
    logger.info(f"Fetching {self.n_files} files to {dest_dir} ({'pooled' if pool else 'sequential'})")

    errors: list[Exception] = []
    counts = {"fetched": 0, "skipped": 0, "promoted": 0}
    n_failed = 0
    total_bytes = 0
    fetched_bytes = 0
    t0 = last_progress = time.monotonic()
    # ``closing`` guarantees the generator's ``finally`` runs even when the loop
    # exits early via ``raise`` (fail-fast) — in pooled mode that ``finally`` is
    # what cancels the still-queued futures so "fail fast" actually stops the run.
    # Tripped the moment a fetch fails, so queued work stops before the unwind
    # reaches the generator's teardown. See ``_attempts``.
    abort = threading.Event()
    attempts = self._attempts(self._iter_attempts(items, dest_dir, do_overwrite), pool, abort)
    try:
        with closing(attempts):
            for item, run in attempts:
                try:
                    status, n_bytes = run()
                except Exception as e:
                    # Tag the exception with the item it came from so a caller
                    # inspecting the ExceptionGroup (or a bare re-raise) can tell
                    # which file failed without cross-referencing logs.
                    e.add_note(f"while fetching {item.rel_path!r} from {item.source_path!r}")
                    n_failed += 1
                    if not continue_on_error:
                        abort.set()
                        raise
                    logger.exception(f"Failed to fetch {item.rel_path}")
                    errors.append(e)
                else:
                    counts[status] += 1
                    total_bytes += n_bytes
                    if status == "fetched":
                        fetched_bytes += n_bytes
                now = time.monotonic()
                if now - last_progress >= _PROGRESS_INTERVAL_S:
                    n_done = sum(counts.values()) + n_failed
                    logger.info(
                        f"Progress: {n_done}/{self.n_files} files "
                        f"({total_bytes / 2**20:.0f} MiB) in {now - t0:.0f}s "
                        f"({total_bytes / 2**20 / max(now - t0, 1e-9):.1f} MiB/s)"
                    )
                    last_progress = now
    finally:
        # Emitted in a ``finally`` so a fail-fast exit still reports the partial
        # totals a multi-hour run accumulated before the failure.
        elapsed = time.monotonic() - t0
        fetched_mib = fetched_bytes / 2**20
        logger.info(
            f"{type(self).__name__}: {counts['fetched']} fetched "
            f"({fetched_mib:.1f} MiB in {elapsed:.1f}s, "
            f"{fetched_mib / max(elapsed, 1e-9):.1f} MiB/s), "
            f"{counts['skipped']} skipped, {counts['promoted']} promoted, "
            f"{n_failed} failed of {self.n_files} files -> {dest_dir}"
        )
    if errors:
        raise ExceptionGroup(f"{len(errors)} of {self.n_files} files failed to download", errors)

source_from_config(cfg)

Construct one :class:Source from a single sources: list entry.

Each entry is a dict carrying at minimum a type: field. Remaining keys are forwarded to the backend’s constructor; the type: key is validated against :data:_SOURCE_TYPES and stripped before forwarding.

Examples:

>>> source_from_config({"type": "http", "urls": ["https://example.com/x.csv"]})
...
<MEDS_extract.download.backends.http.HTTPSource object at 0x...>
>>> source_from_config({"type": "fsspec", "root": "/tmp"})
...
<MEDS_extract.download.backends.fsspec.FsspecSource object at 0x...>
>>> source_from_config(
...     {"type": "physionet", "base_url": "https://physionet.org/files/mimic-iv-demo/2.2"}
... )
<MEDS_extract.download.backends.physionet.PhysioNetSource object at 0x...>

Unknown types surface a clear error:

>>> source_from_config({"type": "s3"})
Traceback (most recent call last):
    ...
ValueError: Unknown source type 's3'. Supported: ['fsspec', 'http', 'physionet'].

Missing type: is flagged the same way. Only the key names are echoed — by the time an entry reaches this function its ${oc.env:...} interpolations are resolved, so echoing values could put live credentials in logs:

>>> source_from_config({"urls": ["https://example.com/x.csv"], "password": "hunter2"})
Traceback (most recent call last):
    ...
ValueError: Source config is missing a 'type:' key. Got keys: ['password', 'urls']

Backend-specific kwargs pass through verbatim — e.g. HTTPSource’s headers: for API-key auth (DataVerse X-Dataverse-key, bearer tokens, Accept:):

>>> src = source_from_config({
...     "type": "http",
...     "urls": ["https://example.com/x.csv"],
...     "headers": {"X-Dataverse-key": "secret-token"},
... })
>>> src._client.headers["X-Dataverse-key"]
'secret-token'
>>> src.close()

That includes the generic include: / exclude: manifest filters every backend accepts:

>>> src = source_from_config({"type": "fsspec", "root": "/tmp", "include": ["hosp/*"]})
>>> src._include
['hosp/*']
Source code in MEDS_extract/download/spec.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def source_from_config(cfg: dict) -> Source:
    """Construct one :class:`Source` from a single ``sources:`` list entry.

    Each entry is a dict carrying at minimum a ``type:`` field. Remaining keys are
    forwarded to the backend's constructor; the ``type:`` key is validated against
    :data:`_SOURCE_TYPES` and stripped before forwarding.

    Examples:
        >>> source_from_config({"type": "http", "urls": ["https://example.com/x.csv"]})
        ... # doctest: +ELLIPSIS
        <MEDS_extract.download.backends.http.HTTPSource object at 0x...>
        >>> source_from_config({"type": "fsspec", "root": "/tmp"})
        ... # doctest: +ELLIPSIS
        <MEDS_extract.download.backends.fsspec.FsspecSource object at 0x...>
        >>> source_from_config(
        ...     {"type": "physionet", "base_url": "https://physionet.org/files/mimic-iv-demo/2.2"}
        ... )  # doctest: +ELLIPSIS
        <MEDS_extract.download.backends.physionet.PhysioNetSource object at 0x...>

        Unknown types surface a clear error:

        >>> source_from_config({"type": "s3"})
        Traceback (most recent call last):
            ...
        ValueError: Unknown source type 's3'. Supported: ['fsspec', 'http', 'physionet'].

        Missing ``type:`` is flagged the same way. Only the key names are echoed —
        by the time an entry reaches this function its ``${oc.env:...}`` interpolations
        are resolved, so echoing values could put live credentials in logs:

        >>> source_from_config({"urls": ["https://example.com/x.csv"], "password": "hunter2"})
        Traceback (most recent call last):
            ...
        ValueError: Source config is missing a 'type:' key. Got keys: ['password', 'urls']

        Backend-specific kwargs pass through verbatim — e.g. ``HTTPSource``'s ``headers:``
        for API-key auth (DataVerse ``X-Dataverse-key``, bearer tokens, ``Accept:``):

        >>> src = source_from_config({
        ...     "type": "http",
        ...     "urls": ["https://example.com/x.csv"],
        ...     "headers": {"X-Dataverse-key": "secret-token"},
        ... })
        >>> src._client.headers["X-Dataverse-key"]
        'secret-token'
        >>> src.close()

        That includes the generic ``include:`` / ``exclude:`` manifest filters every
        backend accepts:

        >>> src = source_from_config({"type": "fsspec", "root": "/tmp", "include": ["hosp/*"]})
        >>> src._include
        ['hosp/*']
    """
    cfg = dict(cfg)
    source_type = cfg.pop("type", None)
    if source_type is None:
        # Echo key names only: values may hold resolved credentials, and this message
        # lands on stderr and in the persisted Hydra log.
        raise ValueError(f"Source config is missing a 'type:' key. Got keys: {sorted(cfg)}")
    if source_type not in _SOURCE_TYPES:
        raise ValueError(f"Unknown source type {source_type!r}. Supported: {sorted(_SOURCE_TYPES)}.")

    module_name, class_name = _SOURCE_TYPES[source_type]
    module = importlib.import_module(module_name, package=__package__)
    return getattr(module, class_name)(**cfg)

sources_from_spec(spec, key='dataset')

Read a full MESSY sources: block and return the configured + common sources.

The key argument selects which bucket of sources to pull — "dataset" (the default), "demo" (for demo downloads), or any other top-level key under sources:. The "common" bucket is always appended and carries shared metadata files (e.g. MIMIC’s concept_map CSVs from GitHub) that all ETL runs need regardless of which primary source is selected.

Examples:

>>> spec = {
...     "sources": {
...         "dataset": [
...             {"type": "http", "urls": ["https://example.com/data.csv"]},
...         ],
...         "demo": [
...             {"type": "http", "urls": ["https://example.com/demo.csv"]},
...         ],
...         "common": [
...             {"type": "http", "urls": ["https://example.com/shared.csv"]},
...         ],
...     },
... }
>>> [type(s).__name__ for s in sources_from_spec(spec, key="dataset")]
['HTTPSource', 'HTTPSource']
>>> [type(s).__name__ for s in sources_from_spec(spec, key="demo")]
['HTTPSource', 'HTTPSource']

Missing keys quietly resolve to an empty list (not an error — a MESSY file that doesn’t declare demo is legal; the CLI layers its own stricter key-must-exist validation on top):

>>> sources_from_spec({"sources": {"dataset": []}}, key="demo")
[]

Missing top-level sources: is also legal (returns empty):

>>> sources_from_spec({}, key="dataset")
[]

key="common" doesn’t double-count — the common bucket is already the selected one, so it isn’t appended a second time:

>>> [type(s).__name__ for s in sources_from_spec(spec, key="common")]
['HTTPSource']

The reserved dataset_version key is metadata, not a bucket — selecting it is an error (its string/mapping value would otherwise be iterated as if it were a source list):

>>> sources_from_spec({"sources": {"dataset_version": "3.1"}}, key="dataset_version")
Traceback (most recent call last):
    ...
ValueError: key='dataset_version' is a reserved sources: key (raw-data version metadata),
not a bucket.
Source code in MEDS_extract/download/spec.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def sources_from_spec(spec: dict, key: str = "dataset") -> list[Source]:
    """Read a full MESSY ``sources:`` block and return the configured + common sources.

    The ``key`` argument selects which bucket of sources to pull — ``"dataset"`` (the
    default), ``"demo"`` (for demo downloads), or any other top-level key under
    ``sources:``. The ``"common"`` bucket is always appended and carries shared
    metadata files (e.g. MIMIC's ``concept_map`` CSVs from GitHub) that all ETL runs
    need regardless of which primary source is selected.

    Examples:
        >>> spec = {
        ...     "sources": {
        ...         "dataset": [
        ...             {"type": "http", "urls": ["https://example.com/data.csv"]},
        ...         ],
        ...         "demo": [
        ...             {"type": "http", "urls": ["https://example.com/demo.csv"]},
        ...         ],
        ...         "common": [
        ...             {"type": "http", "urls": ["https://example.com/shared.csv"]},
        ...         ],
        ...     },
        ... }
        >>> [type(s).__name__ for s in sources_from_spec(spec, key="dataset")]
        ['HTTPSource', 'HTTPSource']
        >>> [type(s).__name__ for s in sources_from_spec(spec, key="demo")]
        ['HTTPSource', 'HTTPSource']

        Missing keys quietly resolve to an empty list (not an error — a MESSY file that
        doesn't declare ``demo`` is legal; the CLI layers its own stricter
        key-must-exist validation on top):

        >>> sources_from_spec({"sources": {"dataset": []}}, key="demo")
        []

        Missing top-level ``sources:`` is also legal (returns empty):

        >>> sources_from_spec({}, key="dataset")
        []

        ``key="common"`` doesn't double-count — the common bucket is already
        the selected one, so it isn't appended a second time:

        >>> [type(s).__name__ for s in sources_from_spec(spec, key="common")]
        ['HTTPSource']

        The reserved ``dataset_version`` key is metadata, not a bucket — selecting
        it is an error (its string/mapping value would otherwise be iterated as if
        it were a source list):

        >>> sources_from_spec({"sources": {"dataset_version": "3.1"}}, key="dataset_version")
        Traceback (most recent call last):
            ...
        ValueError: key='dataset_version' is a reserved sources: key (raw-data version metadata),
        not a bucket.
    """
    if key in SOURCES_RESERVED_KEYS:
        raise ValueError(f"key={key!r} is a reserved sources: key (raw-data version metadata), not a bucket.")
    sources_block = spec.get("sources", {}) or {}
    configured = list(sources_block.get(key, []) or [])
    # ``common`` is always appended UNLESS it's already the selected bucket —
    # otherwise ``sources_from_spec(spec, key="common")`` would build every
    # common backend twice and race two writers on the same dest
    # mid-orchestration.
    common = list(sources_block.get("common", []) or []) if key != "common" else []
    return [source_from_config(c) for c in configured + common]

validate_unique_destinations(sources)

Reject destination collisions across multiple sources sharing one dest_dir.

:attr:Source.files already rejects collisions within one source, but the CLI (and any caller composing sources) stages several sources into one shared directory, where two sources legally listing the same rel_path would race on the same .part file under concurrent workers — or serially clobber each other. Calling this before any fetch turns that late, confusing failure into an immediate, precise config error.

Accessing each source’s :attr:~Source.files materializes its manifest (cached, so the later download_all calls reuse it rather than re-listing).

Examples:

>>> class A(Source):
...     def _list_files(self):
...         return [RemoteFile("x.csv", "")]
...     def _pull(self, source_path, target):
...         target.write_text("A")
>>> class B(Source):
...     def _list_files(self):
...         return [RemoteFile("sub/../x.csv", "")]
...     def _pull(self, source_path, target):
...         target.write_text("B")

Colliding sources are named by their position in the resolved list (plus class name), so several same-type entries — the common case, e.g. a common: bucket of multiple HTTPSource entries — stay distinguishable in the error:

>>> validate_unique_destinations([A(), B()])
Traceback (most recent call last):
    ...
ValueError: Duplicate destination ... 'sub/../x.csv' from B#1 collides with 'x.csv' from A#0.

Distinct destinations pass silently:

>>> class C(Source):
...     def _list_files(self):
...         return [RemoteFile("y.csv", "")]
...     def _pull(self, source_path, target):
...         target.write_text("C")
>>> validate_unique_destinations([A(), C()])
Source code in MEDS_extract/download/source.py 
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
def validate_unique_destinations(sources: Iterable[Source]) -> None:
    """Reject destination collisions across multiple sources sharing one ``dest_dir``.

    :attr:`Source.files` already rejects collisions *within* one source, but the
    CLI (and any caller composing sources) stages several sources into one shared
    directory, where two sources legally listing the same ``rel_path`` would race
    on the same ``.part`` file under concurrent workers — or serially clobber
    each other. Calling this before any fetch turns that late, confusing failure
    into an immediate, precise config error.

    Accessing each source's :attr:`~Source.files` materializes its manifest
    (cached, so the later ``download_all`` calls reuse it rather than re-listing).

    Examples:
        >>> class A(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("x.csv", "")]
        ...     def _pull(self, source_path, target):
        ...         target.write_text("A")
        >>> class B(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("sub/../x.csv", "")]
        ...     def _pull(self, source_path, target):
        ...         target.write_text("B")

        Colliding sources are named by their position in the resolved list (plus
        class name), so several same-type entries — the common case, e.g. a
        ``common:`` bucket of multiple ``HTTPSource`` entries — stay
        distinguishable in the error:

        >>> validate_unique_destinations([A(), B()])
        Traceback (most recent call last):
            ...
        ValueError: Duplicate destination ... 'sub/../x.csv' from B#1 collides with 'x.csv' from A#0.

        Distinct destinations pass silently:

        >>> class C(Source):
        ...     def _list_files(self):
        ...         return [RemoteFile("y.csv", "")]
        ...     def _pull(self, source_path, target):
        ...         target.write_text("C")
        >>> validate_unique_destinations([A(), C()])
    """
    seen: dict[str, tuple[str, RemoteFile]] = {}
    for idx, source in enumerate(sources):
        # ``ClassName#index`` (enumeration order in the resolved list): specs
        # routinely declare several entries of the same type, so the class name
        # alone would leave "HTTPSource collides with HTTPSource" ambiguous.
        name = f"{type(source).__name__}#{idx}"
        for item in source.files:
            if item.dest_key in seen:
                prior_name, prior_item = seen[item.dest_key]
                raise ValueError(
                    f"Duplicate destination across sources: {item.rel_path!r} from {name} "
                    f"collides with {prior_item.rel_path!r} from {prior_name}."
                )
            seen[item.dest_key] = (name, item)