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.

Checking a bundle without importing

validate_bundle runs every check the import runs, without writing. Without a cursor it needs no database, which suits a submission form; with one it also compares the bundle with what is already stored.

from noxdb import init_pool, transaction
from noxdb._import import load_project_dir, validate_bundle

bundle = load_project_dir("IBD_Vienna/")
result = validate_bundle(bundle)                # offline checks only

init_pool()
with transaction() as cur:                      # nothing is written
    result = validate_bundle(bundle, cur=cur, force=True)

for problem in result.errors:
    print("error:", problem)
for note in result.warnings:
    print("warning:", note)

A bundle can also be built in memory from ProjectMeta, SubjectRow, VisitRow, SampleRow and FileRow instead of read from a folder.

Adding to an existing project

A bundle that adds to an existing project (force=True) usually repeats subjects, visits and samples the database already has. The import reuses those rows as they are, so the bundle must agree with them:

• A different value for a stored field (a subject's sex, a visit's age or group, a sample's visit, type, plate coordinates, barcodes, library or antibody class) is an error, and the import is refused. Correct the bundle, or update the database explicitly first.
• A value for a field the database has empty (an age, a sample's SQR or barcodes) is a warning: the import does not fill it in. A sample's SQR / SQRP and barcodes can be filled in with scripts/apply_run_sheet.py (see run_sheet).
• A metadata value that changes is a warning: the import overwrites it.
• An empty value in the bundle asserts nothing.

Top-level entry point

_import

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

Public surface:

• :func:runner.import_project_from_dir validates and imports a folder. The :mod:scripts.import_project CLI is a thin wrapper around it.
• :func:runner.validate_bundle runs the same checks without writing, on a bundle read by :func:loader.load_project_dir or built in memory from the row dataclasses.

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.

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 [])

ValidationResult dataclass

ValidationResult(errors: list[str] = list(), warnings: list[str] = list())

What validate_bundle found.

errors block an import; warnings don't, but are worth reading.

ok property

ok: bool

True when there are no errors.

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

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 (see validate_bundle), so a failed import never leaves the database in a half-written state. With force=True a re-run on the same folder, or a bundle adding to an existing project, 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). A reused subject, visit or sample must carry the same values as the database, or the import is refused.

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 (see
    [`validate_bundle`][noxdb._import.runner.validate_bundle]), so a
    failed import never leaves the database in a half-written state.
    With ``force=True`` a re-run on the same folder, or a bundle adding
    to an existing project, 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``). A reused subject, visit or sample must carry the
    same values as the database, or the import is refused.

    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,
    )

    # The database checks run in their own short read transaction, so
    # every validation error is presented before deciding whether to
    # commit or refuse.
    with transaction() as cur:
        result = validate_bundle(
            bundle, cur=cur, force=force, skip_disk_check=skip_disk_check,
        )
    report.warnings = result.warnings
    errors = result.errors

    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

validate_bundle

validate_bundle(bundle: ProjectBundle, *, cur=None, force: bool = False, skip_disk_check: bool = False) -> ValidationResult

Check a bundle without writing anything.

Every problem is collected, not just the first. Without cur only the checks that need no database run: allowed values, references between the files, duplicates and (unless skip_disk_check) whether the manifest's files exist. That is enough for a form that has no database access.

With cur the bundle is also checked against the database: the project must not exist yet unless force is set, a file path must not belong to another sample, and every subject, visit and sample the database already holds must agree with the bundle (see _validate_existing). The import runs exactly these checks before it commits.

Parameters:

Name	Type	Description	Default
bundle	ProjectBundle	A bundle from load_project_dir, or one built in memory.	required
cur		Cursor to read the database with, e.g. from transaction. Nothing is written.	None
force	bool	Allow adding to a project that already exists.	False
skip_disk_check	bool	Don't check that manifest files exist on this host.	False

Returns:

Type	Description
ValidationResult	A ValidationResult.

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

def validate_bundle(
    bundle: loader.ProjectBundle,
    *,
    cur=None,
    force: bool = False,
    skip_disk_check: bool = False,
) -> ValidationResult:
    """Check a bundle without writing anything.

    Every problem is collected, not just the first. Without *cur* only
    the checks that need no database run: allowed values, references
    between the files, duplicates and (unless *skip_disk_check*) whether
    the manifest's files exist. That is enough for a form that has no
    database access.

    With *cur* the bundle is also checked against the database: the
    project must not exist yet unless *force* is set, a file path must
    not belong to another sample, and every subject, visit and sample
    the database already holds must agree with the bundle (see
    ``_validate_existing``). The import runs exactly these checks before
    it commits.

    Args:
        bundle: A bundle from
            [`load_project_dir`][noxdb._import.loader.load_project_dir],
            or one built in memory.
        cur: Cursor to read the database with, e.g. from
            [`transaction`][noxdb.connection.transaction]. Nothing is written.
        force: Allow adding to a project that already exists.
        skip_disk_check: Don't check that manifest files exist on this host.

    Returns:
        A [`ValidationResult`][noxdb._import.runner.ValidationResult].
    """
    result = ValidationResult(warnings=list(bundle.warnings) + _plate_warnings(bundle))
    result.errors.extend(_validate_schema(bundle))
    result.errors.extend(_validate_referential(bundle))
    if not skip_disk_check:
        result.errors.extend(_validate_disk(bundle))
    if cur is None:
        return result

    existing_project = projects.get_by_name(cur, bundle.project.project_name)
    if existing_project is not None and not force:
        result.errors.append(
            f"project {bundle.project.project_name!r} already exists "
            f"(project_id={existing_project['project_id']}); pass force=True "
            "(--force on the command line) to add to it."
        )
    result.errors.extend(_validate_db_collisions(cur, bundle))
    errs, warns = _validate_existing(cur, bundle)
    result.errors.extend(errs)
    result.warnings.extend(warns)
    return result

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:

1. 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.
2. 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 [])

ValidationResult dataclass

ValidationResult(errors: list[str] = list(), warnings: list[str] = list())

What validate_bundle found.

errors block an import; warnings don't, but are worth reading.

ok property

ok: bool

True when there are no errors.

validate_bundle

validate_bundle(bundle: ProjectBundle, *, cur=None, force: bool = False, skip_disk_check: bool = False) -> ValidationResult

Check a bundle without writing anything.

Every problem is collected, not just the first. Without cur only the checks that need no database run: allowed values, references between the files, duplicates and (unless skip_disk_check) whether the manifest's files exist. That is enough for a form that has no database access.

With cur the bundle is also checked against the database: the project must not exist yet unless force is set, a file path must not belong to another sample, and every subject, visit and sample the database already holds must agree with the bundle (see _validate_existing). The import runs exactly these checks before it commits.

Parameters:

Name	Type	Description	Default
bundle	ProjectBundle	A bundle from load_project_dir, or one built in memory.	required
cur		Cursor to read the database with, e.g. from transaction. Nothing is written.	None
force	bool	Allow adding to a project that already exists.	False
skip_disk_check	bool	Don't check that manifest files exist on this host.	False

Returns:

Type	Description
ValidationResult	A ValidationResult.

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

def validate_bundle(
    bundle: loader.ProjectBundle,
    *,
    cur=None,
    force: bool = False,
    skip_disk_check: bool = False,
) -> ValidationResult:
    """Check a bundle without writing anything.

    Every problem is collected, not just the first. Without *cur* only
    the checks that need no database run: allowed values, references
    between the files, duplicates and (unless *skip_disk_check*) whether
    the manifest's files exist. That is enough for a form that has no
    database access.

    With *cur* the bundle is also checked against the database: the
    project must not exist yet unless *force* is set, a file path must
    not belong to another sample, and every subject, visit and sample
    the database already holds must agree with the bundle (see
    ``_validate_existing``). The import runs exactly these checks before
    it commits.

    Args:
        bundle: A bundle from
            [`load_project_dir`][noxdb._import.loader.load_project_dir],
            or one built in memory.
        cur: Cursor to read the database with, e.g. from
            [`transaction`][noxdb.connection.transaction]. Nothing is written.
        force: Allow adding to a project that already exists.
        skip_disk_check: Don't check that manifest files exist on this host.

    Returns:
        A [`ValidationResult`][noxdb._import.runner.ValidationResult].
    """
    result = ValidationResult(warnings=list(bundle.warnings) + _plate_warnings(bundle))
    result.errors.extend(_validate_schema(bundle))
    result.errors.extend(_validate_referential(bundle))
    if not skip_disk_check:
        result.errors.extend(_validate_disk(bundle))
    if cur is None:
        return result

    existing_project = projects.get_by_name(cur, bundle.project.project_name)
    if existing_project is not None and not force:
        result.errors.append(
            f"project {bundle.project.project_name!r} already exists "
            f"(project_id={existing_project['project_id']}); pass force=True "
            "(--force on the command line) to add to it."
        )
    result.errors.extend(_validate_db_collisions(cur, bundle))
    errs, warns = _validate_existing(cur, bundle)
    result.errors.extend(errs)
    result.warnings.extend(warns)
    return result

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 (see validate_bundle), so a failed import never leaves the database in a half-written state. With force=True a re-run on the same folder, or a bundle adding to an existing project, 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). A reused subject, visit or sample must carry the same values as the database, or the import is refused.

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 (see
    [`validate_bundle`][noxdb._import.runner.validate_bundle]), so a
    failed import never leaves the database in a half-written state.
    With ``force=True`` a re-run on the same folder, or a bundle adding
    to an existing project, 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``). A reused subject, visit or sample must carry the
    same values as the database, or the import is refused.

    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,
    )

    # The database checks run in their own short read transaction, so
    # every validation error is presented before deciding whether to
    # commit or refuse.
    with transaction() as cur:
        result = validate_bundle(
            bundle, cur=cur, force=force, skip_disk_check=skip_disk_check,
        )
    report.warnings = result.warnings
    errors = result.errors

    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

barcodes

barcodes(row: dict[str, str]) -> dict[str, str | None]

The optional barcode cells of a samples row, stripped; empty → None.

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

def barcodes(row: dict[str, str]) -> dict[str, str | None]:
    """The optional barcode cells of a samples row, stripped; empty → ``None``."""
    return {col: (row.get(col) or "").strip() or None for col in schema.BARCODE_COLUMNS}

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 IP or sequencing coordinate cell for import.

Uses the same canonicalization as samples.create, so what the importer accepts here is byte-identical to what gets stored and coordinate 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 IP or sequencing coordinate 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 and coordinate
    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

barcode_errors

barcode_errors(values: dict[str, str | None], *, where: str) -> list[str]

Problems with a samples row's barcode cells.

Parameters:

Name	Type	Description	Default
values	dict[str, str | None]	The row's BARCODE_COLUMNS cells, None when empty.	required
where	str	Location to prefix each message with, e.g. "samples.csv row 4".	required

Returns:

Type	Description
list[str]	One message per index sequence with characters other than
list[str]	A, C, G, T and N, and per value too long for its column.

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

def barcode_errors(values: dict[str, str | None], *, where: str) -> list[str]:
    """Problems with a samples row's barcode cells.

    Args:
        values: The row's ``BARCODE_COLUMNS`` cells, ``None`` when empty.
        where: Location to prefix each message with, e.g.
            ``"samples.csv row 4"``.

    Returns:
        One message per index sequence with characters other than
        A, C, G, T and N, and per value too long for its column.
    """
    errs: list[str] = []
    for column in BARCODE_COLUMNS:
        raw = values.get(column)
        if raw is None:
            continue
        is_id = column.endswith("_id")
        if not is_id:
            try:
                canonical_index(raw)
            except ValueError as exc:
                errs.append(f"{where}.{column}: {exc}")
                continue
        limit = _INDEX_ID_MAX_LEN if is_id else _INDEX_MAX_LEN
        if len(raw.strip()) > limit:
            errs.append(f"{where}.{column}: {raw!r} is longer than {limit} characters")
    return errs

check_no_subject_metadata

check_no_subject_metadata(header: list[str], filename: str) -> None

Refuse meta_* columns in a subjects CSV.

There is no subject-level metadata table, so these columns used to be dropped without a word. Subject-level facts (a diagnosis, a score) belong in visits.csv, which stores them per visit.

Parameters:

Name	Type	Description	Default
header	list[str]	Column names from the CSV header row.	required
filename	str	File name to report in the error.	required

Raises:

Type	Description
ValueError	If any column starts with meta_.

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

def check_no_subject_metadata(header: list[str], filename: str) -> None:
    """Refuse ``meta_*`` columns in a subjects CSV.

    There is no subject-level metadata table, so these columns used to be
    dropped without a word. Subject-level facts (a diagnosis, a score)
    belong in ``visits.csv``, which stores them per visit.

    Args:
        header: Column names from the CSV header row.
        filename: File name to report in the error.

    Raises:
        ValueError: If any column starts with ``meta_``.
    """
    meta = [col for col in header if col.startswith(META_PREFIX)]
    if meta:
        raise ValueError(
            f"{filename}: metadata columns {meta} are not accepted; noxDB "
            "has no subject-level metadata. Move them to visits.csv, which "
            "stores metadata per visit."
        )

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