Skip to content

cli

meds-extract-download — CLI entry point for the download layer.

Reads a MESSY spec’s sources: block and runs each resolved source’s :meth:~MEDS_extract.download.source.Source.download_all in sequence. Sources are processed one at a time; per-file fetches within a source share one :class:~concurrent.futures.ThreadPoolExecutor sized to the user’s concurrency= argument, so the per-file transport bound is a global cap.

Written as a Hydra entry point so override syntax matches the rest of the pipeline. To re-run against a local mirror instead of the original remote, edit the spec’s sources: block (or keep a second bucket — e.g. a mirror: bucket with a type: fsspec entry — and select it via key=mirror); the source definitions themselves live in the spec file, not in Hydra’s config.

The config schema is a hydra_registered_dataclass (from MEDS-transforms): it registers with Hydra’s ConfigStore and types as a dataclass, so cfg.spec / cfg.do_overwrite / etc are typed attributes.

DownloadConfig

Typed config for meds-extract-download.

Fields

spec: Path or pkg:// reference to the MESSY spec YAML with a sources: block. output_dir: Destination directory under which fetched files land. key: Which sources: bucket to pull. "common" is always appended. Must name a bucket that actually exists in the spec (guards against typos silently downloading nothing). concurrency: Max parallel transport streams across all sources (one shared pool). continue_on_error: If True, per-file failures don’t sink the run; every source is still attempted and the process exits non-zero at the end if anything failed. If False (default), the first failing source stops the whole run. do_overwrite: If True, re-fetch every file even if the local copy matches.

Source code in MEDS_extract/download/cli.py 
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
@hydra_registered_dataclass(group=None, name="download_defaults")
class DownloadConfig:
    """Typed config for ``meds-extract-download``.

    Fields:
        spec: Path or ``pkg://`` reference to the MESSY spec YAML with a
            ``sources:`` block.
        output_dir: Destination directory under which fetched files land.
        key: Which ``sources:`` bucket to pull. ``"common"`` is always appended.
            Must name a bucket that actually exists in the spec (guards against
            typos silently downloading nothing).
        concurrency: Max parallel transport streams across all sources (one shared pool).
        continue_on_error: If ``True``, per-file failures don't sink the run; every
            source is still attempted and the process exits non-zero at the end if
            anything failed. If ``False`` (default), the first failing source stops
            the whole run.
        do_overwrite: If ``True``, re-fetch every file even if the local copy matches.
    """

    spec: str = MISSING
    output_dir: str = MISSING
    key: str = "dataset"
    concurrency: int = 4
    continue_on_error: bool = False
    do_overwrite: bool = False

_hydra_main(cfg)

Hydra task function for meds-extract-download; see :func:main.

Required args (Hydra dotlist syntax):

  • spec=/path/to/event_configs.yaml — the MESSY spec with a sources: block (a pkg:// reference also works).
  • output_dir=/path/to/output — where the fetched files land

Optional args:

  • key=dataset (default) / key=demo — 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 is a legitimately download-free ETL and warns + exits 0 regardless of key.
  • concurrency=4 (default) — max parallel transport streams across all sources. One :class:~concurrent.futures.ThreadPoolExecutor is shared by every source’s download_all call so the bound applies globally rather than per-source.
  • continue_on_error=False (default) — if True, per-file failures don’t sink the run and every source is attempted; if False, the first failing source stops the whole run.
  • do_overwrite=False (default) — if True, re-fetch every file even if the local copy matches the manifest.

Exits 0 on full success and 1 on any failure, via an explicit :func:sys.exit — Hydra discards the task function’s return value, so a plain return 1 would not reach the process exit code.

Source code in MEDS_extract/download/cli.py 
 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
@hydra.main(version_base=None, config_path=".", config_name="_cli")
def _hydra_main(cfg: DictConfig) -> None:
    """Hydra task function for ``meds-extract-download``; see :func:`main`.

    Required args (Hydra dotlist syntax):

    - ``spec=/path/to/event_configs.yaml`` — the MESSY spec with a ``sources:`` block
      (a ``pkg://`` reference also works).
    - ``output_dir=/path/to/output`` — where the fetched files land

    Optional args:

    - ``key=dataset`` (default) / ``key=demo`` — 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 is a legitimately download-free ETL and warns +
      exits 0 regardless of ``key``.
    - ``concurrency=4`` (default) — max parallel transport streams across all sources.
      One :class:`~concurrent.futures.ThreadPoolExecutor` is shared by every source's
      ``download_all`` call so the bound applies globally rather than per-source.
    - ``continue_on_error=False`` (default) — if True, per-file failures don't sink the
      run and every source is attempted; if False, the first failing source stops the
      whole run.
    - ``do_overwrite=False`` (default) — if True, re-fetch every file even if the local
      copy matches the manifest.

    Exits ``0`` on full success and ``1`` on any failure, via an explicit
    :func:`sys.exit` — Hydra discards the task function's *return* value, so a
    plain ``return 1`` would not reach the process exit code.
    """
    # One load of the one config object, then everything comes off it. Errors —
    # spec resolution, a URL-shaped output_dir, document validation, bad bucket key,
    # malformed source entries, unresolvable interpolations in the selected bucket —
    # are user config mistakes: log them and exit 1, mirroring the manifest-validation
    # handling below, rather than dumping a raw traceback through Hydra.
    try:
        output_dir = user_local_path(str(cfg.output_dir), field="output_dir")
        messy = MessyConfig.load(str(cfg.spec))
        sources = messy.selected_sources(key=cfg.key)
    except (TypeError, ValueError, FileNotFoundError) as e:
        logger.error(f"Could not construct sources from the spec: {e}")
        sys.exit(1)

    if not sources:
        logger.warning(f"No sources resolved for key={cfg.key!r} in {messy.source_fp}. Nothing to do.")
        return

    # Teardown notes:
    #
    # - ``shutdown(wait=False, cancel_futures=True)`` cancels *queued* futures
    #   immediately. Worker threads are NOT daemon threads (since Python 3.9,
    #   bpo-39812), so the interpreter joins any still-running workers at exit —
    #   in-flight transfers finish (or die when their transport is torn down)
    #   before the process can exit.
    # - The ExitStack closes sources LIFO *before* the pool-shutdown callback runs,
    #   so on any exit path each owned ``httpx.Client`` is closed while workers may
    #   still be streaming — those in-flight HTTP transfers fail fast rather than
    #   draining, which is what keeps Ctrl+C reasonably prompt for HTTP sources.
    #   Fsspec copies have no equivalent abort path and run to completion.
    with ExitStack() as stack:
        pool = ThreadPoolExecutor(max_workers=cfg.concurrency)
        stack.callback(pool.shutdown, wait=False, cancel_futures=True)
        for source in sources:
            stack.enter_context(source)

        # Materializes every source's manifest up-front (cached for the fetch loop
        # below) and fails before any fetch into output_dir if a manifest row is
        # malformed or two sources would write the same file. (Manifest listing
        # itself may do network I/O — e.g. PhysioNet's SHA256SUMS.txt GET, fsspec
        # source-side hashing.)
        try:
            validate_unique_destinations(sources)
        except ValueError as e:
            logger.error(f"Source manifests failed validation: {e}")
            sys.exit(1)

        all_ok = True
        for source in sources:
            try:
                source.download_all(
                    output_dir,
                    pool=pool,
                    continue_on_error=cfg.continue_on_error,
                    do_overwrite=cfg.do_overwrite,
                )
            except Exception:
                logger.exception(f"download_all failed for {type(source).__name__}")
                all_ok = False
                if not cfg.continue_on_error:
                    # Fail fast applies across sources too: don't start source N+1
                    # after source N has already sunk the run.
                    break
        if not all_ok:
            sys.exit(1)

main()

Console-script entry point for meds-extract-download.

Validates the required dotlist args before Hydra owns the process: the Hydra run dir is anchored at ${output_dir}/..., so without this check a bare invocation would die inside interpolation resolution instead of printing usage — and a failed invocation must create no directories anywhere.

Source code in MEDS_extract/download/cli.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def main() -> None:
    """Console-script entry point for ``meds-extract-download``.

    Validates the required dotlist args before Hydra owns the process: the Hydra run
    dir is anchored at ``${output_dir}/...``, so without this check a bare invocation
    would die inside interpolation resolution instead of printing usage — and a failed
    invocation must create no directories anywhere.
    """
    require_dotlist_args(
        "meds-extract-download",
        {"spec": "<pkg://...|/path|./path>", "output_dir": "<dir>"},
        local_only=("output_dir",),
    )
    _hydra_main()