Project import¶

The master importer loads a whole project folder (project.yaml, subjects.csv, visits.csv, samples.csv, files/manifest.csv) into the database in a single transaction. See the CLI page for the command-line wrapper.

Top-level entry point¶

_import ¶

Master import subpackage: project-folder → database in one transaction.

Public surface is :func:runner.import_project_from_dir. The :mod:scripts.import_project CLI is a thin wrapper around it.

ProjectImportError ¶

ProjectImportError(message: str, errors: list[str] | None = None)

Bases: RuntimeError

Raised when validation fails or the project already exists without --force.

Source code in src/noxdb/_import/runner.py

def __init__(self, message: str, errors: list[str] | None = None) -> None:
    super().__init__(message)
    self.errors = list(errors or [])

import_project_from_dir ¶

import_project_from_dir(root: str | Path, *, dry_run: bool = False, force: bool = False, compute_md5: bool = False, skip_disk_check: bool = False, log_dir: str | Path | None = None) -> ImportReport

Validate and (unless dry_run) import the project under root.

The runner performs validation in a read-only pass before any writes happen, so a failed import never leaves the database in a half-written state. With force=True a re-run on the same folder re-uses existing rows via the get_or_create / get_or_register / set_* semantics of the CRUD layer; the report distinguishes inserted from existing (or, for metadata, inserted vs updated vs unchanged).

Cross-project collisions on sample_name or file_path are refused even with force=True — those UNIQUEs are global by design.

Parameters:

Name	Type	Description	Default
`root`	`str \| Path`	Path to the project folder containing `project.yaml`, `subjects.csv`, `visits.csv`, `samples.csv`, and `files/manifest.csv`.	required
`dry_run`	`bool`	Validate only; skip the commit phase.	`False`
`force`	`bool`	Allow re-import of a project that already exists.	`False`
`compute_md5`	`bool`	Hash files whose manifest entry has no `checksum_md5`.	`False`
`skip_disk_check`	`bool`	Skip the per-file `os.path.exists` check (use when files live on a remote mount not visible from this host).	`False`
`log_dir`	`str \| Path \| None`	Directory to write the JSON report to. Defaults to `~/.noxdb/imports/`.	`None`

Returns:

Type	Description
`ImportReport`	An `ImportReport` with row counts per table.

Raises:

Type	Description
`ProjectImportError`	With the collected error list when validation fails, or when the project already exists and `force` is `False`.

Source code in src/noxdb/_import/runner.py

def import_project_from_dir(
    root: str | Path,
    *,
    dry_run: bool = False,
    force: bool = False,
    compute_md5: bool = False,
    skip_disk_check: bool = False,
    log_dir: str | Path | None = None,
) -> ImportReport:
    """Validate and (unless *dry_run*) import the project under *root*.

    The runner performs validation in a read-only pass before any
    writes happen, so a failed import never leaves the database in a
    half-written state. With ``force=True`` a re-run on the same
    folder re-uses existing rows via the
    ``get_or_create`` / ``get_or_register`` / ``set_*`` semantics of
    the CRUD layer; the report distinguishes ``inserted`` from
    ``existing`` (or, for metadata, ``inserted`` vs ``updated`` vs
    ``unchanged``).

    Cross-project collisions on ``sample_name`` or ``file_path`` are
    refused even with ``force=True`` — those UNIQUEs are global by
    design.

    Args:
        root: Path to the project folder containing ``project.yaml``,
            ``subjects.csv``, ``visits.csv``, ``samples.csv``, and
            ``files/manifest.csv``.
        dry_run: Validate only; skip the commit phase.
        force: Allow re-import of a project that already exists.
        compute_md5: Hash files whose manifest entry has no
            ``checksum_md5``.
        skip_disk_check: Skip the per-file ``os.path.exists`` check
            (use when files live on a remote mount not visible from
            this host).
        log_dir: Directory to write the JSON report to. Defaults to
            ``~/.noxdb/imports/``.

    Returns:
        An `ImportReport` with row counts per table.

    Raises:
        ProjectImportError: With the collected error list when
            validation fails, or when the project already exists and
            ``force`` is ``False``.
    """
    start = time.monotonic()
    bundle = loader.load_project_dir(root)
    report = ImportReport(
        project_name=bundle.project.project_name,
        dry_run=dry_run,
        force=force,
        warnings=list(bundle.warnings) + _plate_warnings(bundle),
    )

    errors: list[str] = []
    errors.extend(_validate_schema(bundle))
    errors.extend(_validate_referential(bundle))
    if not skip_disk_check:
        errors.extend(_validate_disk(bundle))

    # The collision check needs a cursor; run it in its own short read
    # transaction so we can present all validation errors before deciding
    # whether to commit or refuse.
    with transaction() as cur:
        existing_project = projects.get_by_name(cur, bundle.project.project_name)
        if existing_project is not None and not force:
            errors.append(
                f"project {bundle.project.project_name!r} already exists "
                f"(project_id={existing_project['project_id']}); rerun with "
                "force=True to append."
            )
        errors.extend(_validate_db_collisions(cur, bundle))

    if errors:
        report.errors = errors
        report.duration_seconds = time.monotonic() - start
        _write_log(log_dir, report)
        raise ProjectImportError(
            f"import refused: {len(errors)} validation error(s)", errors,
        )

    if dry_run:
        report.duration_seconds = time.monotonic() - start
        _write_log(log_dir, report)
        return report

    with transaction() as cur:
        counts, pid = _commit(cur, bundle, compute_md5=compute_md5, skip_disk_check=skip_disk_check)
    report.counts = counts
    report.project_id = pid
    report.duration_seconds = time.monotonic() - start
    _write_log(log_dir, report)
    return report

Runner¶

runner ¶

Validate a :class:ProjectBundle and (optionally) commit it.

The runner is split into two distinct phases so import errors never leave the database in a half-written state:

Validation — schema, referential, duplicate, on-disk path, and project-existence checks. Errors are collected exhaustively (not short-circuit) so the user sees every problem in one pass.
Commit — a single :func:transaction block calling the existing CRUD wrappers in hierarchical order. An exception anywhere rolls back the entire import; partial states are impossible.

The split also gives --dry-run for free: skip phase 2.

ProjectImportError ¶

ProjectImportError(message: str, errors: list[str] | None = None)

Bases: RuntimeError

Raised when validation fails or the project already exists without --force.

Source code in src/noxdb/_import/runner.py

def __init__(self, message: str, errors: list[str] | None = None) -> None:
    super().__init__(message)
    self.errors = list(errors or [])

import_project_from_dir ¶

import_project_from_dir(root: str | Path, *, dry_run: bool = False, force: bool = False, compute_md5: bool = False, skip_disk_check: bool = False, log_dir: str | Path | None = None) -> ImportReport

Validate and (unless dry_run) import the project under root.

The runner performs validation in a read-only pass before any writes happen, so a failed import never leaves the database in a half-written state. With force=True a re-run on the same folder re-uses existing rows via the get_or_create / get_or_register / set_* semantics of the CRUD layer; the report distinguishes inserted from existing (or, for metadata, inserted vs updated vs unchanged).

Cross-project collisions on sample_name or file_path are refused even with force=True — those UNIQUEs are global by design.

Parameters:

Name	Type	Description	Default
`root`	`str \| Path`	Path to the project folder containing `project.yaml`, `subjects.csv`, `visits.csv`, `samples.csv`, and `files/manifest.csv`.	required
`dry_run`	`bool`	Validate only; skip the commit phase.	`False`
`force`	`bool`	Allow re-import of a project that already exists.	`False`
`compute_md5`	`bool`	Hash files whose manifest entry has no `checksum_md5`.	`False`
`skip_disk_check`	`bool`	Skip the per-file `os.path.exists` check (use when files live on a remote mount not visible from this host).	`False`
`log_dir`	`str \| Path \| None`	Directory to write the JSON report to. Defaults to `~/.noxdb/imports/`.	`None`

Returns:

Type	Description
`ImportReport`	An `ImportReport` with row counts per table.

Raises:

Type	Description
`ProjectImportError`	With the collected error list when validation fails, or when the project already exists and `force` is `False`.

Source code in src/noxdb/_import/runner.py

def import_project_from_dir(
    root: str | Path,
    *,
    dry_run: bool = False,
    force: bool = False,
    compute_md5: bool = False,
    skip_disk_check: bool = False,
    log_dir: str | Path | None = None,
) -> ImportReport:
    """Validate and (unless *dry_run*) import the project under *root*.

    The runner performs validation in a read-only pass before any
    writes happen, so a failed import never leaves the database in a
    half-written state. With ``force=True`` a re-run on the same
    folder re-uses existing rows via the
    ``get_or_create`` / ``get_or_register`` / ``set_*`` semantics of
    the CRUD layer; the report distinguishes ``inserted`` from
    ``existing`` (or, for metadata, ``inserted`` vs ``updated`` vs
    ``unchanged``).

    Cross-project collisions on ``sample_name`` or ``file_path`` are
    refused even with ``force=True`` — those UNIQUEs are global by
    design.

    Args:
        root: Path to the project folder containing ``project.yaml``,
            ``subjects.csv``, ``visits.csv``, ``samples.csv``, and
            ``files/manifest.csv``.
        dry_run: Validate only; skip the commit phase.
        force: Allow re-import of a project that already exists.
        compute_md5: Hash files whose manifest entry has no
            ``checksum_md5``.
        skip_disk_check: Skip the per-file ``os.path.exists`` check
            (use when files live on a remote mount not visible from
            this host).
        log_dir: Directory to write the JSON report to. Defaults to
            ``~/.noxdb/imports/``.

    Returns:
        An `ImportReport` with row counts per table.

    Raises:
        ProjectImportError: With the collected error list when
            validation fails, or when the project already exists and
            ``force`` is ``False``.
    """
    start = time.monotonic()
    bundle = loader.load_project_dir(root)
    report = ImportReport(
        project_name=bundle.project.project_name,
        dry_run=dry_run,
        force=force,
        warnings=list(bundle.warnings) + _plate_warnings(bundle),
    )

    errors: list[str] = []
    errors.extend(_validate_schema(bundle))
    errors.extend(_validate_referential(bundle))
    if not skip_disk_check:
        errors.extend(_validate_disk(bundle))

    # The collision check needs a cursor; run it in its own short read
    # transaction so we can present all validation errors before deciding
    # whether to commit or refuse.
    with transaction() as cur:
        existing_project = projects.get_by_name(cur, bundle.project.project_name)
        if existing_project is not None and not force:
            errors.append(
                f"project {bundle.project.project_name!r} already exists "
                f"(project_id={existing_project['project_id']}); rerun with "
                "force=True to append."
            )
        errors.extend(_validate_db_collisions(cur, bundle))

    if errors:
        report.errors = errors
        report.duration_seconds = time.monotonic() - start
        _write_log(log_dir, report)
        raise ProjectImportError(
            f"import refused: {len(errors)} validation error(s)", errors,
        )

    if dry_run:
        report.duration_seconds = time.monotonic() - start
        _write_log(log_dir, report)
        return report

    with transaction() as cur:
        counts, pid = _commit(cur, bundle, compute_md5=compute_md5, skip_disk_check=skip_disk_check)
    report.counts = counts
    report.project_id = pid
    report.duration_seconds = time.monotonic() - start
    _write_log(log_dir, report)
    return report

Loader¶

loader ¶

Read a project folder into typed in-memory records.

The runner consumes the structures produced here and performs validation + commit. Loading itself is forgiving — it does not check enum values or referential integrity, only structural things (required columns present, file readable, YAML parseable).

ProjectBundle `dataclass` ¶

ProjectBundle(root: Path, project: ProjectMeta, subjects: list[SubjectRow] = list(), visits: list[VisitRow] = list(), samples: list[SampleRow] = list(), files: list[FileRow] = list(), warnings: list[str] = list())

Everything read from a project folder, pre-validation.

load_project_dir ¶

load_project_dir(root: str | Path) -> ProjectBundle

Read all required files from a project folder.

Loading is forgiving — does not check enum values or referential integrity, only structural things (required columns present, file readable, YAML parseable).

Parameters:

Name	Type	Description	Default
`root`	`str \| Path`	Path to the project folder.	required

Returns:

Type	Description
`ProjectBundle`	A
`ProjectBundle`	`ProjectBundle`
`ProjectBundle`	with the parsed contents. Unknown columns produce warnings
`ProjectBundle`	stored on the bundle rather than raising.

Raises:

Type	Description
`FileNotFoundError`	For missing required files.
`ValueError`	For missing required columns.

Source code in src/noxdb/_import/loader.py

def load_project_dir(root: str | Path) -> ProjectBundle:
    """Read all required files from a project folder.

    Loading is forgiving — does not check enum values or referential
    integrity, only structural things (required columns present, file
    readable, YAML parseable).

    Args:
        root: Path to the project folder.

    Returns:
        A
        [`ProjectBundle`][noxdb._import.loader.ProjectBundle]
        with the parsed contents. Unknown columns produce warnings
        stored on the bundle rather than raising.

    Raises:
        FileNotFoundError: For missing required files.
        ValueError: For missing required columns.
    """
    root_path = Path(root)
    if not root_path.is_dir():
        raise FileNotFoundError(f"project directory not found: {root_path}")

    bundle = ProjectBundle(
        root=root_path,
        project=_load_project_yaml(root_path / "project.yaml"),
    )
    bundle.subjects, sub_warn = _load_subjects(root_path / "subjects.csv")
    bundle.visits, vis_warn = _load_visits(root_path / "visits.csv")
    bundle.samples, sam_warn = _load_samples(root_path / "samples.csv")
    bundle.files, fil_warn = _load_manifest(root_path / "files" / "manifest.csv")
    bundle.warnings = sub_warn + vis_warn + sam_warn + fil_warn
    return bundle

Schema¶

schema ¶

CSV / YAML schema declarations and value coercion helpers.

Each CSV file has a fixed set of required columns plus optional ones; any column that starts with meta_ is treated as a typed metadata key (the prefix is stripped to produce the stored key_name). Type inference order: int → float → bool ('true'/'false'/'1'/'0') → str. Empty cells are treated as "no metadata for this row" (no INSERT).

coerce_metadata_value ¶

coerce_metadata_value(raw: str) -> Any | None

Coerce a raw CSV cell to an int / float / bool / str.

Type inference order: bool ('true' / 'false') → int → float → str. '1' / '0' are NOT treated as bools (they'd otherwise parse as int and lose their boolean intent at write time).

Parameters:

Name	Type	Description	Default
`raw`	`str`	The raw cell value.	required

Returns:

Type	Description
`Any \| None`	The coerced value, or `None` for empty / whitespace-only
`Any \| None`	cells (meaning "no metadata entry" — the caller should skip
`Any \| None`	the row rather than insert NULL, because
`Any \| None`	`metadata.set_visit` /
`Any \| None`	`metadata.set_sample`
`Any \| None`	reject `None`).

Source code in src/noxdb/_import/schema.py

def coerce_metadata_value(raw: str) -> Any | None:
    """Coerce a raw CSV cell to an ``int`` / ``float`` / ``bool`` / ``str``.

    Type inference order: bool (``'true'`` / ``'false'``) → int → float
    → str. ``'1'`` / ``'0'`` are NOT treated as bools (they'd otherwise
    parse as int and lose their boolean intent at write time).

    Args:
        raw: The raw cell value.

    Returns:
        The coerced value, or ``None`` for empty / whitespace-only
        cells (meaning "no metadata entry" — the caller should skip
        the row rather than insert NULL, because
        [`metadata.set_visit`][noxdb.metadata.set_visit] /
        [`metadata.set_sample`][noxdb.metadata.set_sample]
        reject ``None``).
    """
    if raw is None:
        return None
    s = raw.strip()
    if s == "":
        return None
    # bool BEFORE int: '1' and '0' would otherwise parse as int.
    low = s.lower()
    if low in ("true", "false"):
        return low == "true"
    # int
    try:
        return int(s)
    except ValueError:
        pass
    # float
    try:
        return float(s)
    except ValueError:
        pass
    return s

coerce_int ¶

coerce_int(raw: str, *, field: str) -> int

Parse an int with a labelled error on failure.

Parameters:

Name	Type	Description	Default
`raw`	`str`	The raw cell value.	required
`field`	`str`	Human-readable identifier (e.g. `"visits.csv row 4.age"`) embedded in the error message so the user can locate the bad cell.	required

Returns:

Type	Description
`int`	The parsed integer.

Raises:

Type	Description
`ValueError`	If `raw` is not parseable as an int.

Source code in src/noxdb/_import/schema.py

def coerce_int(raw: str, *, field: str) -> int:
    """Parse an int with a labelled error on failure.

    Args:
        raw: The raw cell value.
        field: Human-readable identifier (e.g. ``"visits.csv row 4.age"``)
            embedded in the error message so the user can locate the
            bad cell.

    Returns:
        The parsed integer.

    Raises:
        ValueError: If ``raw`` is not parseable as an int.
    """
    try:
        return int(str(raw).strip())
    except (TypeError, ValueError) as exc:
        raise ValueError(f"{field}: expected int, got {raw!r}") from exc

validate_plate_id ¶

validate_plate_id(raw: str | None, *, field: str) -> tuple[str, str | None]

Validate + canonicalize an SQR / SQRP cell for import.

Uses the same canonicalization as samples.create, so what the importer accepts here is byte-identical to what gets stored — SQR+SQRP plate matching can't drift between the two.

Parameters:

Name	Type	Description	Default
`raw`	`str \| None`	The raw cell value.	required
`field`	`str`	Human-readable identifier (e.g. `"samples.csv row 4.sqr"`) embedded in messages so the user can locate the cell.	required

Returns:

Type	Description
`str`	`(canonical, warning)` — canonical is the value that will
`str \| None`	be stored; warning is a human-readable string when
`tuple[str, str \| None]`	canonicalization changed the input (whitespace stripped or an
`tuple[str, str \| None]`	`NA`/empty sentinel collapsed), else `None`.

Raises:

Type	Description
`ValueError`	If the canonical value exceeds the `samples.SQR` / `samples.SQRP` column width (10), which would otherwise fail with a cryptic driver error mid-commit.

Source code in src/noxdb/_import/schema.py

def validate_plate_id(raw: str | None, *, field: str) -> tuple[str, str | None]:
    """Validate + canonicalize an SQR / SQRP cell for import.

    Uses the same canonicalization as
    [`samples.create`][noxdb.samples.create], so what the importer
    accepts here is byte-identical to what gets stored — SQR+SQRP
    plate matching can't drift between the two.

    Args:
        raw: The raw cell value.
        field: Human-readable identifier (e.g.
            ``"samples.csv row 4.sqr"``) embedded in messages so the
            user can locate the cell.

    Returns:
        ``(canonical, warning)`` — *canonical* is the value that will
        be stored; *warning* is a human-readable string when
        canonicalization changed the input (whitespace stripped or an
        ``NA``/empty sentinel collapsed), else ``None``.

    Raises:
        ValueError: If the canonical value exceeds the
            ``samples.SQR`` / ``samples.SQRP`` column width (10),
            which would otherwise fail with a cryptic driver error
            mid-commit.
    """
    canon = canonical_plate_id(raw)
    if len(canon) > _PLATE_MAX_LEN:
        raise ValueError(
            f"{field}: {raw!r} is {len(canon)} chars after normalization; "
            f"max is {_PLATE_MAX_LEN}"
        )
    warning = None
    if canon != (raw or "").strip():
        warning = f"{field}: {raw!r} normalized to {canon!r}"
    return canon, warning

split_columns ¶

split_columns(header: list[str], required: tuple[str, ...], optional: tuple[str, ...]) -> tuple[list[str], list[str]]

Split a CSV header into known, metadata, and extra columns.

Parameters:

Name	Type	Description	Default
`header`	`list[str]`	List of column names from the CSV header row.	required
`required`	`tuple[str, ...]`	Column names that must be present.	required
`optional`	`tuple[str, ...]`	Column names allowed but not required.	required

Returns:

Type	Description
`list[str]`	`(non_meta_extra, meta_keys)`:
`list[str]`	`non_meta_extra` — columns not in required/optional and not prefixed with `meta_`. These are silently ignored by the loader but reported to the user as warnings (so a typo like `smaple_name` doesn't silently drop data).
`tuple[list[str], list[str]]`	`meta_keys` — the bare metadata key names with the `meta_` prefix stripped.

Source code in src/noxdb/_import/schema.py

def split_columns(
    header: list[str], required: tuple[str, ...], optional: tuple[str, ...],
) -> tuple[list[str], list[str]]:
    """Split a CSV header into known, metadata, and extra columns.

    Args:
        header: List of column names from the CSV header row.
        required: Column names that must be present.
        optional: Column names allowed but not required.

    Returns:
        ``(non_meta_extra, meta_keys)``:

        - ``non_meta_extra`` — columns not in required/optional and
          not prefixed with ``meta_``. These are silently ignored by
          the loader but reported to the user as warnings (so a typo
          like ``smaple_name`` doesn't silently drop data).
        - ``meta_keys`` — the bare metadata key names with the
          ``meta_`` prefix stripped.
    """
    known = set(required) | set(optional)
    extra: list[str] = []
    meta: list[str] = []
    for col in header:
        if col in known:
            continue
        if col.startswith(META_PREFIX):
            key = col[len(META_PREFIX):]
            if key:
                meta.append(key)
        else:
            extra.append(col)
    return extra, meta

Project import¶

Top-level entry point¶

_import ¶

ProjectImportError ¶

import_project_from_dir ¶

Runner¶

runner ¶

ProjectImportError ¶

import_project_from_dir ¶

Loader¶

loader ¶

ProjectBundle dataclass ¶

load_project_dir ¶

Schema¶

schema ¶

coerce_metadata_value ¶

coerce_int ¶

validate_plate_id ¶

split_columns ¶

ProjectBundle `dataclass` ¶