fetch

fetch

Export project metadata + files from the database to a local folder.

These helpers are the "consumer" side of noxdb: given a project_id, materialize either the metadata table (CSV / Excel) or the file payloads (downloaded via SFTP when the database is reached through an SSH jump host, or copied directly from the filesystem when running on LiSC).

Layout produced by :func:export_project:

<output_dir>/
├── metadata.csv                # tidy wide-form table
├── metadata.xlsx               # same, Excel
├── README.txt                  # project summary
└── files/
    └── <sample_name>/<file_type>.<ext>     # layout='by_sample'
    # or
    └── <file_type>/<sample_name>.<ext>     # layout='by_type'

A file registered as a member of a tar (archive_member set) is extracted on its own: read at archive_offset when known, otherwise looked up by name in the tar, and checked against checksum_md5.

All functions take an optional cursor and open their own :func:noxdb.transaction block when one isn't provided, so they work as one-shot calls from a notebook or composed inside a larger read transaction.

export_metadata_table

export_metadata_table(cur=None, *, project_id: int, output_dir: Path | str, formats: tuple[str, ...] = ('csv',)) -> dict[str, Path]

Write the project tidy table to disk in the requested formats.

Parameters:

Name	Type	Description	Default
cur		Optional cursor from transaction(). When None this helper opens its own transaction.	None
project_id	int	Project to export.	required
output_dir	Path | str	Destination directory. Created if missing.	required
formats	tuple[str, ...]	Iterable of 'csv' and/or 'xlsx'.	('csv',)

Returns:

Type	Description
dict[str, Path]	{fmt: Path, ...} mapping each requested format to the file
dict[str, Path]	written.

Raises:

Type	Description
ValueError	If formats contains an unknown format.
ImportError	If 'xlsx' is requested but openpyxl is missing, or if pandas is missing entirely.

Source code in src/noxdb/fetch.py

def export_metadata_table(
    cur=None,
    *,
    project_id: int,
    output_dir: Path | str,
    formats: tuple[str, ...] = ("csv",),
) -> dict[str, Path]:
    """Write the project tidy table to disk in the requested formats.

    Args:
        cur: Optional cursor from `transaction()`. When ``None`` this
            helper opens its own transaction.
        project_id: Project to export.
        output_dir: Destination directory. Created if missing.
        formats: Iterable of ``'csv'`` and/or ``'xlsx'``.

    Returns:
        ``{fmt: Path, ...}`` mapping each requested format to the file
        written.

    Raises:
        ValueError: If ``formats`` contains an unknown format.
        ImportError: If ``'xlsx'`` is requested but ``openpyxl`` is
            missing, or if pandas is missing entirely.
    """
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    with _cur_ctx(cur) as c:
        df = queries.project_tidy_table(c, project_id)

    written: dict[str, Path] = {}
    for fmt in formats:
        if fmt == "csv":
            path = out / "metadata.csv"
            df.to_csv(path, index=False)
            written["csv"] = path
        elif fmt == "xlsx":
            path = out / "metadata.xlsx"
            df.to_excel(path, index=False)
            written["xlsx"] = path
        else:
            raise ValueError(f"unknown format {fmt!r}; expected 'csv' or 'xlsx'")
    return written

download_files_for_project

download_files_for_project(cur=None, *, project_id: int, output_dir: Path | str, file_types: list[str] | None = None, layout: str = 'by_sample', config_path: str | Path | None = DEFAULT_CONFIG_PATH, ssh_section: str = DEFAULT_SSH_SECTION, **ssh_overrides: Any) -> dict[str, Any]

Copy every file registered for a project into a local folder.

SSH credentials are resolved from config_path / ssh_section the same way init_pool does. When ssh_host is unset the function falls back to local file copy via shutil.copyfile (useful when running on LiSC itself). Already-present destinations are skipped, so the call is resumable.

Parameters:

Name	Type	Description	Default
cur		Optional cursor from transaction(). When None this helper opens its own transaction.	None
project_id	int	Project to download.	required
output_dir	Path | str	Destination directory. Created if missing.	required
file_types	list[str] | None	Only download files whose file_type is in the list. None means every type.	None
layout	str	'by_sample' groups files under per-sample subdirectories (<sample>/<file_type>.<ext>); 'by_type' groups by file_type (<file_type>/<sample>.<ext>); 'flat' writes every file at the top level (note: file_path is globally UNIQUE but basenames are not). A tar member is always saved under its own name inside the tar, so the runs of a resequenced sample don't overwrite each other.	'by_sample'
config_path	str | Path | None	Path to the MariaDB-style config file. None disables config-file lookup.	DEFAULT_CONFIG_PATH
ssh_section	str	Section name within config_path to read SSH credentials from.	DEFAULT_SSH_SECTION
**ssh_overrides	Any	Per-call SSH kwargs that win over the config file and NOXDB_SSH_* env vars.	{}

Returns:

Type	Description
dict[str, Any]	``{"downloaded": [...], "skipped": [...], "failed": [...],
dict[str, Any]	"output_dir": str}. Every entry hasfile_path``,
dict[str, Any]	archive_member and dst; downloaded entries include
dict[str, Any]	size, failed entries include error.

Raises:

Type	Description
ValueError	Unknown layout.
ImportError	If SFTP is needed and paramiko is missing.

Source code in src/noxdb/fetch.py

def download_files_for_project(
    cur=None,
    *,
    project_id: int,
    output_dir: Path | str,
    file_types: list[str] | None = None,
    layout: str = "by_sample",
    config_path: str | Path | None = DEFAULT_CONFIG_PATH,
    ssh_section: str = DEFAULT_SSH_SECTION,
    **ssh_overrides: Any,
) -> dict[str, Any]:
    """Copy every file registered for a project into a local folder.

    SSH credentials are resolved from ``config_path`` / ``ssh_section``
    the same way
    [`init_pool`][noxdb.connection.init_pool] does. When
    ``ssh_host`` is unset the function falls back to local file copy
    via ``shutil.copyfile`` (useful when running on LiSC itself).
    Already-present destinations are skipped, so the call is resumable.

    Args:
        cur: Optional cursor from `transaction()`. When ``None`` this
            helper opens its own transaction.
        project_id: Project to download.
        output_dir: Destination directory. Created if missing.
        file_types: Only download files whose ``file_type`` is in the
            list. ``None`` means every type.
        layout: ``'by_sample'`` groups files under per-sample
            subdirectories (``<sample>/<file_type>.<ext>``); ``'by_type'``
            groups by file_type (``<file_type>/<sample>.<ext>``);
            ``'flat'`` writes every file at the top level (note:
            ``file_path`` is globally UNIQUE but basenames are not). A tar
            member is always saved under its own name inside the tar, so
            the runs of a resequenced sample don't overwrite each other.
        config_path: Path to the MariaDB-style config file. ``None``
            disables config-file lookup.
        ssh_section: Section name within ``config_path`` to read SSH
            credentials from.
        **ssh_overrides: Per-call SSH kwargs that win over the config
            file and ``NOXDB_SSH_*`` env vars.

    Returns:
        ``{"downloaded": [...], "skipped": [...], "failed": [...],
        "output_dir": str}``. Every entry has ``file_path``,
        ``archive_member`` and ``dst``; ``downloaded`` entries include
        ``size``, ``failed`` entries include ``error``.

    Raises:
        ValueError: Unknown ``layout``.
        ImportError: If SFTP is needed and ``paramiko`` is missing.
    """
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)
    with _cur_ctx(cur) as c:
        df = queries.files_for_project(c, project_id)

    if df.empty:
        return {
            "downloaded": [],
            "skipped": [],
            "failed": [],
            "output_dir": str(out),
        }

    if file_types is not None:
        df = df[df["file_type"].isin(file_types)]

    creds = _ssh_credentials(
        config_path=config_path, section=ssh_section, **ssh_overrides
    )
    downloaded: list[dict[str, Any]] = []
    skipped: list[dict[str, Any]] = []
    failed: list[dict[str, Any]] = []

    with _Transport(creds) as transport:
        for row in df.to_dict("records"):
            member = row["archive_member"]
            dst = _layout_target(
                layout, out, row["sample_name"], row["file_type"], row["file_path"], member,
            )
            entry = {"file_path": row["file_path"], "archive_member": member, "dst": str(dst)}
            if dst.exists():
                skipped.append(entry)
                continue
            try:
                if member:
                    size = transport.fetch_member(
                        row["file_path"], member, dst,
                        offset=_opt_int(row["archive_offset"]),
                        size=_opt_int(row["file_size_bytes"]),
                        md5=row["checksum_md5"] if isinstance(row["checksum_md5"], str) else None,
                    )
                else:
                    size = transport.fetch(row["file_path"], dst)
            except Exception as exc:
                failed.append({**entry, "error": str(exc)})
                continue
            downloaded.append({**entry, "size": size})

    return {
        "downloaded": downloaded,
        "skipped": skipped,
        "failed": failed,
        "output_dir": str(out),
    }

export_project

export_project(cur=None, *, project_id: int, output_dir: Path | str, file_types: list[str] | None = None, layout: str = 'by_sample', metadata_formats: tuple[str, ...] = ('csv',), include_files: bool = True, config_path: str | Path | None = DEFAULT_CONFIG_PATH, ssh_section: str = DEFAULT_SSH_SECTION, **ssh_overrides: Any) -> dict[str, Any]

One-shot export: metadata table + files + README.

Combines export_metadata_table and download_files_for_project and writes a small README.txt describing the project. Useful for handing a self-contained snapshot to a collaborator.

Parameters:

Name	Type	Description	Default
cur		Optional cursor from transaction(). When None this helper opens its own transaction.	None
project_id	int	Project to export.	required
output_dir	Path | str	Destination directory. Created if missing.	required
file_types	list[str] | None	Forwarded to download_files_for_project.	None
layout	str	Forwarded to download_files_for_project.	'by_sample'
metadata_formats	tuple[str, ...]	Forwarded to export_metadata_table.	('csv',)
include_files	bool	When False, skip file download entirely (only metadata + README are produced).	True
config_path	str | Path | None	See download_files_for_project.	DEFAULT_CONFIG_PATH
ssh_section	str	See download_files_for_project.	DEFAULT_SSH_SECTION
**ssh_overrides	Any	See download_files_for_project.	{}

Returns:

Type	Description
dict[str, Any]	``{"project", "summary", "metadata", "files", "readme",
dict[str, Any]	"output_dir"}`` — the project row, the
dict[str, Any]	project_summary
dict[str, Any]	dict, paths of the metadata files, the file-download report,
dict[str, Any]	and the README path.

Source code in src/noxdb/fetch.py

def export_project(
    cur=None,
    *,
    project_id: int,
    output_dir: Path | str,
    file_types: list[str] | None = None,
    layout: str = "by_sample",
    metadata_formats: tuple[str, ...] = ("csv",),
    include_files: bool = True,
    config_path: str | Path | None = DEFAULT_CONFIG_PATH,
    ssh_section: str = DEFAULT_SSH_SECTION,
    **ssh_overrides: Any,
) -> dict[str, Any]:
    """One-shot export: metadata table + files + README.

    Combines
    [`export_metadata_table`][noxdb.fetch.export_metadata_table]
    and
    [`download_files_for_project`][noxdb.fetch.download_files_for_project]
    and writes a small ``README.txt`` describing the project. Useful
    for handing a self-contained snapshot to a collaborator.

    Args:
        cur: Optional cursor from `transaction()`. When ``None`` this
            helper opens its own transaction.
        project_id: Project to export.
        output_dir: Destination directory. Created if missing.
        file_types: Forwarded to
            [`download_files_for_project`][noxdb.fetch.download_files_for_project].
        layout: Forwarded to
            [`download_files_for_project`][noxdb.fetch.download_files_for_project].
        metadata_formats: Forwarded to
            [`export_metadata_table`][noxdb.fetch.export_metadata_table].
        include_files: When ``False``, skip file download entirely
            (only metadata + README are produced).
        config_path: See
            [`download_files_for_project`][noxdb.fetch.download_files_for_project].
        ssh_section: See
            [`download_files_for_project`][noxdb.fetch.download_files_for_project].
        **ssh_overrides: See
            [`download_files_for_project`][noxdb.fetch.download_files_for_project].

    Returns:
        ``{"project", "summary", "metadata", "files", "readme",
        "output_dir"}`` — the project row, the
        [`project_summary`][noxdb.queries.project_summary]
        dict, paths of the metadata files, the file-download report,
        and the README path.
    """
    out = Path(output_dir)
    out.mkdir(parents=True, exist_ok=True)

    with _cur_ctx(cur) as c:
        project_row = projects.get(c, project_id)
        summary = queries.project_summary(c, project_id)
        meta_paths = export_metadata_table(
            c,
            project_id=project_id,
            output_dir=out,
            formats=metadata_formats,
        )

    file_report: dict[str, Any] = {
        "downloaded": [], "skipped": [], "failed": [], "output_dir": None,
    }
    if include_files:
        file_report = download_files_for_project(
            cur=cur,
            project_id=project_id,
            output_dir=out / "files",
            file_types=file_types,
            layout=layout,
            config_path=config_path,
            ssh_section=ssh_section,
            **ssh_overrides,
        )

    readme_path = out / "README.txt"
    readme_path.write_text(_render_readme(project_row, summary), encoding="utf-8")

    return {
        "project": project_row,
        "summary": summary,
        "metadata": {fmt: str(p) for fmt, p in meta_paths.items()},
        "files": file_report,
        "readme": str(readme_path),
        "output_dir": str(out),
    }