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           -> archive
counts / beer_norm / zigp_norm / edger_norm /
zigp_loose / fastq_qc                              -> 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.

Files inside a tar

A row can point at one member of a tar archive (schema 006): file_path is the .tar, archive_member the member's name inside it and archive_offset the byte position of its data. Size and checksum then describe the member and must come from the caller, since the member cannot be stat'ed. Plain files have archive_member = ''.

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, archive_member: str = '', archive_offset: int | None = None, file_size_bytes: int | None = None) -> 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, zigp_loose).	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
skip_disk_check	bool	Skip the path-prefix and on-disk checks.	False
archive_member	str	Name of the file inside the tar at file_path. Empty (default) for a plain file.	''
archive_offset	int | None	Byte position of the member's data in the tar. Only with archive_member.	None
file_size_bytes	int | None	The member's size. Only with archive_member; a plain file's size is read from disk.	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. With archive_member: file_path not a .tar or compute_md5 set; without it: archive_offset or file_size_bytes 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, archive_member) (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,
    archive_member: str = "",
    archive_offset: int | None = None,
    file_size_bytes: int | None = None,
) -> 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`, `zigp_loose`).
        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.
        skip_disk_check: Skip the path-prefix and on-disk checks.
        archive_member: Name of the file inside the tar at
            ``file_path``. Empty (default) for a plain file.
        archive_offset: Byte position of the member's data in the tar.
            Only with ``archive_member``.
        file_size_bytes: The member's size. Only with ``archive_member``;
            a plain file's size is read from disk.

    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. With
            ``archive_member``: ``file_path`` not a ``.tar`` or
            ``compute_md5`` set; without it: ``archive_offset`` or
            ``file_size_bytes`` 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``, ``archive_member``) (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,
        archive_member=archive_member,
        archive_offset=archive_offset,
        file_size_bytes=file_size_bytes,
    )
    cur.execute(
        "INSERT INTO sample_files "
        "(sample_id, file_type, file_path, archive_member, archive_offset, "
        "file_size_bytes, checksum_md5, storage_tier) "
        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        (
            sample_id,
            row["file_type"],
            row["file_path"],
            row["archive_member"],
            row["archive_offset"],
            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, archive_member: str = '', archive_offset: int | None = None, file_size_bytes: int | None = None) -> tuple[int, bool]

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

If a row with this file_path and archive_member 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
skip_disk_check	bool	Used only on insert.	False
archive_member	str	Part of the lookup key; see register.	''
archive_offset	int | None	Used only on insert.	None
file_size_bytes	int | 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,
    archive_member: str = "",
    archive_offset: int | None = None,
    file_size_bytes: int | None = None,
) -> tuple[int, bool]:
    """Idempotently register a file. Returns ``(file_id, registered)``.

    If a row with this ``file_path`` and ``archive_member`` 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.
        skip_disk_check: Used only on insert.
        archive_member: Part of the lookup key; see
            [`register`][noxdb.files.register].
        archive_offset: Used only on insert.
        file_size_bytes: 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, archive_member)
    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,
            archive_member=archive_member,
            archive_offset=archive_offset,
            file_size_bytes=file_size_bytes,
        )
    except mariadb.IntegrityError:
        existing = get_by_path(cur, file_path, archive_member)
        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, archive_member: 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 (the tar, for a tar member).	required
archive_member	str	Member name inside the tar; empty for a plain file.	''

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, archive_member: 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 (the tar, for a tar member).
        archive_member: Member name inside the tar; empty for a plain file.

    Returns:
        The row as ``dict[str, Any]``, or ``None`` if not found.
    """
    cur.execute(
        "SELECT * FROM sample_files WHERE file_path = ? AND archive_member = ?",
        (file_path, archive_member),
    )
    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.
ValueError	If the row is a tar member; the tar's size and checksum are not the member's.

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.
        ValueError: If the row is a tar member; the tar's size and
            checksum are not the member's.
    """
    row = get(cur, file_id)
    if row is None:
        return False
    if row["archive_member"]:
        raise ValueError(
            f"file_id {file_id} is a tar member; restat() only applies to plain files"
        )
    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, archive_member: str = '') -> 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
archive_member	str	With path: the member inside the tar; empty for a plain file.	''

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,
    archive_member: str = "",
) -> 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``).
        archive_member: With ``path``: the member inside the tar; empty
            for a plain file.

    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 = ? AND archive_member = ?",
            (path, archive_member),
        )
    return cur.fetchone() is not None