files

files

Registration wrapper for the sample_files table.

Unlike the other CRUD modules, files.register doubles as a gate between application code and the schema: it stats the file on disk, validates that the path lives in the right storage tier, and only then hands the row to MariaDB. The schema's CHECKs (absolute path, MD5 format, UNIQUE path) are still in place but are now a backstop, not the first line of defense.

Storage tier policy

File-type → tier is fixed by lab convention:

fastq_r1 / fastq_r2 / fastq_single / bam / counts  -> archive
beer_norm / zigp_norm / edger_norm                 -> work

Roots are configurable via env vars (defaults shown):

NOXDB_ARCHIVE_ROOT  default /lisc/archive
NOXDB_WORK_ROOT     default /lisc/work

Callers can override storage_tier to 'scratch' or 'external' (escape hatches with no path-prefix check); overriding to swap archive/work against the type-derived value is rejected.

register

register(cur, sample_id: int, file_path: str, file_type: str, *, compute_md5: bool = False, checksum_md5: str | None = None, storage_tier: str | None = None, skip_disk_check: bool = False) -> int

Validate a file on disk and insert a sample_files row.

Filesystem checks (path is absolute, regular file exists, extension matches file_type, path lives under the tier root via realpath) run before any SQL is executed, so a failure leaves the transaction untouched.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
sample_id	int	Parent sample. Must already exist.	required
file_path	str	Absolute path on disk.	required
file_type	str	One of the known types (fastq_r1, fastq_r2, fastq_single, bam, counts, beer_norm, zigp_norm, edger_norm).	required
compute_md5	bool	If True, hash the file. Mutually exclusive with checksum_md5.	False
checksum_md5	str | None	Caller-supplied 32-char lowercase-hex MD5.	None
storage_tier	str | None	Override the file-type-derived tier. Only 'scratch' and 'external' are accepted as overrides; flipping archive ↔ work is rejected.	None

Returns:

Type	Description
int	The newly inserted file_id.

Raises:

Type	Description
ValueError	Relative path, unknown file_type, mismatched extension or tier, malformed checksum_md5, or both compute_md5 and checksum_md5 set.
FileNotFoundError	If the path does not exist.
IsADirectoryError	If the path is a directory.
IntegrityError	Unknown sample_id (FK violation) or duplicate file_path (global UNIQUE).

Source code in src/noxdb/files.py

def register(
    cur,
    sample_id: int,
    file_path: str,
    file_type: str,
    *,
    compute_md5: bool = False,
    checksum_md5: str | None = None,
    storage_tier: str | None = None,
    skip_disk_check: bool = False,
) -> int:
    """Validate a file on disk and insert a `sample_files` row.

    Filesystem checks (path is absolute, regular file exists, extension
    matches ``file_type``, path lives under the tier root via realpath)
    run **before** any SQL is executed, so a failure leaves the
    transaction untouched.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        sample_id: Parent sample. Must already exist.
        file_path: Absolute path on disk.
        file_type: One of the known types (`fastq_r1`, `fastq_r2`,
            `fastq_single`, `bam`, `counts`, `beer_norm`, `zigp_norm`,
            `edger_norm`).
        compute_md5: If ``True``, hash the file. Mutually exclusive with
            ``checksum_md5``.
        checksum_md5: Caller-supplied 32-char lowercase-hex MD5.
        storage_tier: Override the file-type-derived tier. Only
            ``'scratch'`` and ``'external'`` are accepted as overrides;
            flipping ``archive`` ↔ ``work`` is rejected.

    Returns:
        The newly inserted ``file_id``.

    Raises:
        ValueError: Relative path, unknown ``file_type``, mismatched
            extension or tier, malformed ``checksum_md5``, or both
            ``compute_md5`` and ``checksum_md5`` set.
        FileNotFoundError: If the path does not exist.
        IsADirectoryError: If the path is a directory.
        mariadb.IntegrityError: Unknown ``sample_id`` (FK violation) or
            duplicate ``file_path`` (global UNIQUE).
    """
    row = _inspect_file(
        file_path, file_type,
        compute_md5=compute_md5,
        checksum_md5=checksum_md5,
        storage_tier=storage_tier,
        skip_disk_check=skip_disk_check,
    )
    cur.execute(
        "INSERT INTO sample_files "
        "(sample_id, file_type, file_path, file_size_bytes, checksum_md5, "
        "storage_tier) VALUES (?, ?, ?, ?, ?, ?)",
        (
            sample_id,
            row["file_type"],
            row["file_path"],
            row["file_size_bytes"],
            row["checksum_md5"],
            row["storage_tier"],
        ),
    )
    return cur.lastrowid

get_or_register

get_or_register(cur, sample_id: int, file_path: str, file_type: str, *, compute_md5: bool = False, checksum_md5: str | None = None, storage_tier: str | None = None, skip_disk_check: bool = False) -> tuple[int, bool]

Idempotently register a file. Returns (file_id, registered).

If a row with this file_path already exists, it is returned as-is — the file is NOT re-stat'd and the other arguments are not used to update the existing row. This means a stale path that was registered in the past keeps returning its id even if the file has since been deleted; call restat if you need to refresh it.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
sample_id	int	Parent sample (used only on insert).	required
file_path	str	Absolute path on disk. Globally unique.	required
file_type	str	See register.	required
compute_md5	bool	Used only on insert.	False
checksum_md5	str | None	Used only on insert.	None
storage_tier	str | None	Used only on insert.	None

Returns:

Type	Description
int	(file_id, registered) where registered is True iff
bool	this call inserted the row.

Raises:

Type	Description
IntegrityError	If the race-recovery fetch also misses.

Source code in src/noxdb/files.py

def get_or_register(
    cur,
    sample_id: int,
    file_path: str,
    file_type: str,
    *,
    compute_md5: bool = False,
    checksum_md5: str | None = None,
    storage_tier: str | None = None,
    skip_disk_check: bool = False,
) -> tuple[int, bool]:
    """Idempotently register a file. Returns ``(file_id, registered)``.

    If a row with this ``file_path`` already exists, it is returned
    as-is — the file is NOT re-stat'd and the other arguments are not
    used to update the existing row. This means a stale path that was
    registered in the past keeps returning its id even if the file has
    since been deleted; call
    [`restat`][noxdb.files.restat] if you need to refresh it.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        sample_id: Parent sample (used only on insert).
        file_path: Absolute path on disk. Globally unique.
        file_type: See [`register`][noxdb.files.register].
        compute_md5: Used only on insert.
        checksum_md5: Used only on insert.
        storage_tier: Used only on insert.

    Returns:
        ``(file_id, registered)`` where ``registered`` is ``True`` iff
        this call inserted the row.

    Raises:
        mariadb.IntegrityError: If the race-recovery fetch also misses.
        Plus everything [`register`][noxdb.files.register] raises
        on insert.
    """
    existing = get_by_path(cur, file_path)
    if existing is not None:
        return int(existing["file_id"]), False
    try:
        new_id = register(
            cur, sample_id, file_path, file_type,
            compute_md5=compute_md5,
            checksum_md5=checksum_md5,
            storage_tier=storage_tier,
            skip_disk_check=skip_disk_check,
        )
    except mariadb.IntegrityError:
        existing = get_by_path(cur, file_path)
        if existing is None:
            raise
        return int(existing["file_id"]), False
    return new_id, True

get

get(cur, file_id: int) -> dict[str, Any] | None

Return the file row for a given id.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
file_id	int	Primary key to look up.	required

Returns:

Type	Description
dict[str, Any] | None	The row as dict[str, Any], or None if not found.

Source code in src/noxdb/files.py

def get(cur, file_id: int) -> dict[str, Any] | None:
    """Return the file row for a given id.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        file_id: Primary key to look up.

    Returns:
        The row as ``dict[str, Any]``, or ``None`` if not found.
    """
    cur.execute("SELECT * FROM sample_files WHERE file_id = ?", (file_id,))
    row = cur.fetchone()
    return _row_to_dict(cur, row) if row is not None else None

get_by_path

get_by_path(cur, file_path: str) -> dict[str, Any] | None

Return the file row for a given path.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
file_path	str	Absolute path on disk.	required

Returns:

Type	Description
dict[str, Any] | None	The row as dict[str, Any], or None if not found.

Source code in src/noxdb/files.py

def get_by_path(cur, file_path: str) -> dict[str, Any] | None:
    """Return the file row for a given path.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        file_path: Absolute path on disk.

    Returns:
        The row as ``dict[str, Any]``, or ``None`` if not found.
    """
    cur.execute("SELECT * FROM sample_files WHERE file_path = ?", (file_path,))
    row = cur.fetchone()
    return _row_to_dict(cur, row) if row is not None else None

list_for_sample

list_for_sample(cur, sample_id: int, *, order_by: str = 'file_id') -> list[dict[str, Any]]

Return all sample_files rows for a sample.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
sample_id	int	Sample to list.	required
order_by	str	Column name to order by. Must be a column of sample_files.	'file_id'

Returns:

Type	Description
list[dict[str, Any]]	All matching rows as list[dict[str, Any]].

Raises:

Type	Description
ValueError	If order_by is not a known column name.

Source code in src/noxdb/files.py

def list_for_sample(
    cur, sample_id: int, *, order_by: str = "file_id"
) -> list[dict[str, Any]]:
    """Return all ``sample_files`` rows for a sample.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        sample_id: Sample to list.
        order_by: Column name to order by. Must be a column of ``sample_files``.

    Returns:
        All matching rows as ``list[dict[str, Any]]``.

    Raises:
        ValueError: If ``order_by`` is not a known column name.
    """
    if order_by not in _ORDERABLE:
        raise ValueError(
            f"order_by must be one of {sorted(_ORDERABLE)}, got {order_by!r}"
        )
    cur.execute(
        f"SELECT * FROM sample_files WHERE sample_id = ? ORDER BY {order_by}",
        (sample_id,),
    )
    rows = cur.fetchall()
    columns = [d[0] for d in cur.description]
    return [dict(zip(columns, row)) for row in rows]

count_for_sample

count_for_sample(cur, sample_id: int) -> int

Return the number of files registered for a sample.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
sample_id	int	Sample to count.	required

Returns:

Type	Description
int	Number of sample_files rows.

Source code in src/noxdb/files.py

def count_for_sample(cur, sample_id: int) -> int:
    """Return the number of files registered for a sample.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        sample_id: Sample to count.

    Returns:
        Number of ``sample_files`` rows.
    """
    cur.execute(
        "SELECT COUNT(*) FROM sample_files WHERE sample_id = ?", (sample_id,)
    )
    return int(cur.fetchone()[0])

update

update(cur, file_id: int, *, file_size_bytes: int | None = None, checksum_md5: str | None = None, storage_tier: str | None = None) -> bool

Partial update of a file row.

Only kwargs with non-None values are written. file_path, file_type, sample_id, and created_at are NOT updatable — those describe a different file. Use restat to refresh size/checksum from disk after a file is rewritten in place.

Updating storage_tier enforces the same file_type → tier invariant as register: flipping archive ↔ work is rejected; scratch / external overrides are allowed.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
file_id	int	Row to update.	required
file_size_bytes	int | None	New size (if not None).	None
checksum_md5	str | None	New 32-char lowercase-hex MD5 (if not None).	None
storage_tier	str | None	New tier (if not None).	None

Returns:

Type	Description
bool	True iff exactly one row was updated.

Raises:

Type	Description
ValueError	Malformed checksum_md5, unknown storage_tier, or a tier override that violates the file-type invariant.

Source code in src/noxdb/files.py

def update(
    cur,
    file_id: int,
    *,
    file_size_bytes: int | None = None,
    checksum_md5: str | None = None,
    storage_tier: str | None = None,
) -> bool:
    """Partial update of a file row.

    Only kwargs with non-None values are written. ``file_path``,
    ``file_type``, ``sample_id``, and ``created_at`` are NOT updatable
    — those describe a different file. Use
    [`restat`][noxdb.files.restat] to refresh size/checksum from
    disk after a file is rewritten in place.

    Updating ``storage_tier`` enforces the same `file_type → tier`
    invariant as [`register`][noxdb.files.register]: flipping
    ``archive`` ↔ ``work`` is rejected; ``scratch`` / ``external``
    overrides are allowed.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        file_id: Row to update.
        file_size_bytes: New size (if not None).
        checksum_md5: New 32-char lowercase-hex MD5 (if not None).
        storage_tier: New tier (if not None).

    Returns:
        ``True`` iff exactly one row was updated.

    Raises:
        ValueError: Malformed ``checksum_md5``, unknown
            ``storage_tier``, or a tier override that violates the
            file-type invariant.
    """
    if checksum_md5 is not None:
        _validate_md5(checksum_md5)
    if storage_tier is not None:
        if storage_tier not in _ALL_TIERS:
            raise ValueError(
                f"storage_tier must be one of {sorted(_ALL_TIERS)}, got {storage_tier!r}"
            )
        # Enforce the same file_type -> tier invariant as register():
        # callers may move a file to scratch/external, but cannot flip
        # archive <-> work for a given file_type.
        cur.execute(
            "SELECT file_type FROM sample_files WHERE file_id = ?", (file_id,)
        )
        row = cur.fetchone()
        if row is not None:
            _resolve_tier(row[0], storage_tier)
    fields = {
        "file_size_bytes": file_size_bytes,
        "checksum_md5": checksum_md5,
        "storage_tier": storage_tier,
    }
    assignments = [(col, val) for col, val in fields.items() if val is not None]
    if not assignments:
        return False
    set_clause = ", ".join(f"{col} = ?" for col, _ in assignments)
    params = [val for _, val in assignments]
    params.append(file_id)
    cur.execute(
        f"UPDATE sample_files SET {set_clause} WHERE file_id = ?", tuple(params)
    )
    return cur.rowcount > 0

restat

restat(cur, file_id: int, *, compute_md5: bool = False) -> bool

Re-read size (and optionally md5) from disk for an existing row.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
file_id	int	Row to refresh.	required
compute_md5	bool	If True, recompute the MD5. Otherwise the existing checksum is kept.	False

Returns:

Type	Description
bool	True iff the row's columns actually changed.

Raises:

Type	Description
FileNotFoundError	If the path no longer resolves. No SQL runs in that case.

Source code in src/noxdb/files.py

def restat(cur, file_id: int, *, compute_md5: bool = False) -> bool:
    """Re-read size (and optionally md5) from disk for an existing row.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        file_id: Row to refresh.
        compute_md5: If ``True``, recompute the MD5. Otherwise the
            existing checksum is kept.

    Returns:
        ``True`` iff the row's columns actually changed.

    Raises:
        FileNotFoundError: If the path no longer resolves. No SQL runs
            in that case.
    """
    row = get(cur, file_id)
    if row is None:
        return False
    st = _stat_regular_file(row["file_path"])
    new_size = int(st.st_size)
    new_md5 = _compute_md5(row["file_path"]) if compute_md5 else row["checksum_md5"]
    return update(
        cur, file_id,
        file_size_bytes=new_size,
        checksum_md5=new_md5,
    )

delete

delete(cur, file_id: int) -> bool

Delete a file row.

Only removes the database record. The file on disk is untouched — clean it up separately.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
file_id	int	Row to delete.	required

Returns:

Type	Description
bool	True iff a row was removed.

Source code in src/noxdb/files.py

def delete(cur, file_id: int) -> bool:
    """Delete a file row.

    Only removes the database record. The file on disk is untouched —
    clean it up separately.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        file_id: Row to delete.

    Returns:
        ``True`` iff a row was removed.
    """
    cur.execute("DELETE FROM sample_files WHERE file_id = ?", (file_id,))
    return cur.rowcount > 0

exists

exists(cur, file_id: int | None = None, *, path: str | None = None) -> bool

Return whether a file with the given id or path exists.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
file_id	int | None	Id to check (exclusive with path).	None
path	str | None	Path to check (exclusive with file_id).	None

Returns:

Type	Description
bool	True if a matching row exists.

Raises:

Type	Description
ValueError	If both or neither of file_id / path is given.

Source code in src/noxdb/files.py

def exists(
    cur,
    file_id: int | None = None,
    *,
    path: str | None = None,
) -> bool:
    """Return whether a file with the given id or path exists.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        file_id: Id to check (exclusive with ``path``).
        path: Path to check (exclusive with ``file_id``).

    Returns:
        ``True`` if a matching row exists.

    Raises:
        ValueError: If both or neither of ``file_id`` / ``path`` is given.
    """
    if (file_id is None) == (path is None):
        raise ValueError("exists() requires exactly one of file_id or path")
    if file_id is not None:
        cur.execute("SELECT 1 FROM sample_files WHERE file_id = ?", (file_id,))
    else:
        cur.execute("SELECT 1 FROM sample_files WHERE file_path = ?", (path,))
    return cur.fetchone() is not None