subjects

subjects

CRUD wrapper for the subjects table.

Same call style as noxdb.projects — every function takes a cursor first; callers wrap them in with transaction() as cur:.

subject_code is globally unique — subjects carry no project affiliation (project membership lives in project_samples, reached via the sample → visit → subject lineage). The natural-key lookup is get_by_code rather than get.

create

create(cur, subject_code: str, sex: str | None, *, origin: str | None = None) -> int

Insert a subject and return its new subject_id.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
subject_code	str	Stable, globally unique subject code.	required
sex	str | None	'M', 'F', or None for controls without a known sex (DB-side CHECK allows NULL).	required
origin	str | None	Optional free-text origin.	None

Returns:

Type	Description
int	The newly inserted subject_id.

Raises:

Type	Description
IntegrityError	If subject_code already exists (global UNIQUE), or sex is a non-null value outside ('M', 'F'). Use get_or_create for idempotent inserts.

Source code in src/noxdb/subjects.py

def create(
    cur,
    subject_code: str,
    sex: str | None,
    *,
    origin: str | None = None,
) -> int:
    """Insert a subject and return its new ``subject_id``.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        subject_code: Stable, globally unique subject code.
        sex: ``'M'``, ``'F'``, or ``None`` for controls without a
            known sex (DB-side CHECK allows NULL).
        origin: Optional free-text origin.

    Returns:
        The newly inserted ``subject_id``.

    Raises:
        mariadb.IntegrityError: If ``subject_code`` already exists
            (global UNIQUE), or ``sex`` is a non-null value outside
            ``('M', 'F')``. Use
            [`get_or_create`][noxdb.subjects.get_or_create]
            for idempotent inserts.
    """
    cur.execute(
        "INSERT INTO subjects (subject_code, sex, origin) "
        "VALUES (?, ?, ?)",
        (subject_code, sex if sex and sex.upper() not in ("NA", "N/A") else None, origin),
    )
    return cur.lastrowid

get

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

Return the subject row for a given id.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
subject_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/subjects.py

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

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

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

get_by_code

get_by_code(cur, subject_code: str) -> dict[str, Any] | None

Return the subject row for the globally-unique natural key.

Hot path for CSV importers: look up by subject_code before inserting.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
subject_code	str	Globally unique subject code.	required

Returns:

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

Source code in src/noxdb/subjects.py

def get_by_code(cur, subject_code: str) -> dict[str, Any] | None:
    """Return the subject row for the globally-unique natural key.

    Hot path for CSV importers: look up by ``subject_code`` before
    inserting.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        subject_code: Globally unique subject code.

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

get_or_create

get_or_create(cur, subject_code: str, sex: str | None, *, origin: str | None = None) -> tuple[int, bool]

Idempotently return the subject id, inserting if needed.

Existing rows are returned as-is and never updated by this call. sex / origin are still checked against the existing row: a mismatch raises (see Raises) so an accidental cross-study subject_code collision fails loudly instead of silently merging two unrelated subjects. Falls back to a re-fetch on the UNIQUE-violation race where another transaction inserted the same key in parallel.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
subject_code	str	Globally unique subject code.	required
sex	str | None	Used only on insert.	required
origin	str | None	Used only on insert.	None

Returns:

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

Raises:

Type	Description
ValueError	If a subject with this subject_code already exists but with a different non-NULL sex / origin (a likely accidental cross-study collision — see :func:_assert_no_conflict).
IntegrityError	If the race-recovery fetch also misses.

Source code in src/noxdb/subjects.py

def get_or_create(
    cur,
    subject_code: str,
    sex: str | None,
    *,
    origin: str | None = None,
) -> tuple[int, bool]:
    """Idempotently return the subject id, inserting if needed.

    Existing rows are returned as-is and never *updated* by this call.
    *sex* / *origin* are still checked against the existing row: a
    mismatch raises (see Raises) so an accidental cross-study
    ``subject_code`` collision fails loudly instead of silently
    merging two unrelated subjects. Falls back to a re-fetch on the
    UNIQUE-violation race where another transaction inserted the same
    key in parallel.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        subject_code: Globally unique subject code.
        sex: Used only on insert.
        origin: Used only on insert.

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

    Raises:
        ValueError: If a subject with this ``subject_code`` already
            exists but with a different non-NULL ``sex`` / ``origin``
            (a likely accidental cross-study collision — see
            :func:`_assert_no_conflict`).
        mariadb.IntegrityError: If the race-recovery fetch also misses.
    """
    existing = get_by_code(cur, subject_code)
    if existing is not None:
        _assert_no_conflict(existing, sex, origin)
        return int(existing["subject_id"]), False
    try:
        new_id = create(
            cur, subject_code, sex, origin=origin,
        )
        return new_id, True
    except mariadb.IntegrityError:
        existing = get_by_code(cur, subject_code)
        if existing is None:
            raise
        _assert_no_conflict(existing, sex, origin)
        return int(existing["subject_id"]), False

list_for_project

list_for_project(cur, project_id: int, *, order_by: str = 'subject_id') -> list[dict[str, Any]]

Return all subjects with at least one sample in a project.

Project membership lives in project_samples; this traverses project_samples → samples → visits → subjects and de-duplicates, since a subject can have many samples in the same project.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
project_id	int	Project to list.	required
order_by	str	Column name to order by. Must be a column of subjects.	'subject_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/subjects.py

def list_for_project(
    cur, project_id: int, *, order_by: str = "subject_id"
) -> list[dict[str, Any]]:
    """Return all subjects with at least one sample in a project.

    Project membership lives in ``project_samples``; this traverses
    ``project_samples → samples → visits → subjects`` and de-duplicates,
    since a subject can have many samples in the same project.

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

    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(
        "SELECT DISTINCT sub.* FROM project_samples ps "
        "JOIN samples sm   ON sm.sample_id   = ps.sample_id "
        "JOIN visits v     ON v.visit_id     = sm.visit_id "
        "JOIN subjects sub ON sub.subject_id = v.subject_id "
        f"WHERE ps.project_id = ? ORDER BY sub.{order_by}",
        (project_id,),
    )
    rows = cur.fetchall()
    columns = [d[0] for d in cur.description]
    return [dict(zip(columns, row)) for row in rows]

count_for_project

count_for_project(cur, project_id: int) -> int

Return the number of distinct subjects with samples in a project.

Traverses project_samples → samples → visits → subjects and counts distinct subjects (a subject can have many samples in the project).

Parameters:

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

Returns:

Type	Description
int	Number of distinct subject rows.

Source code in src/noxdb/subjects.py

def count_for_project(cur, project_id: int) -> int:
    """Return the number of distinct subjects with samples in a project.

    Traverses ``project_samples → samples → visits → subjects`` and
    counts distinct subjects (a subject can have many samples in the
    project).

    Args:
        cur: Audit-logging cursor from `transaction()`.
        project_id: Project to count.

    Returns:
        Number of distinct subject rows.
    """
    cur.execute(
        "SELECT COUNT(DISTINCT sub.subject_id) FROM project_samples ps "
        "JOIN samples sm   ON sm.sample_id   = ps.sample_id "
        "JOIN visits v     ON v.visit_id     = sm.visit_id "
        "JOIN subjects sub ON sub.subject_id = v.subject_id "
        "WHERE ps.project_id = ?",
        (project_id,),
    )
    return int(cur.fetchone()[0])

update

update(cur, subject_id: int, *, subject_code: str | None = None, sex: str | None = None, origin: str | None = None) -> bool

Partial update of a subject row.

Only kwargs with non-None values are written. created_at is intentionally NOT updatable here. Use raw SQL if you really need it.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
subject_id	int	Row to update.	required
subject_code	str | None	New code (if not None).	None
sex	str | None	New sex (if not None).	None
origin	str | None	New origin (if not None).	None

Returns:

Type	Description
bool	True iff exactly one row was updated. False (no SQL
bool	run) when every kwarg is None.

Source code in src/noxdb/subjects.py

def update(
    cur,
    subject_id: int,
    *,
    subject_code: str | None = None,
    sex: str | None = None,
    origin: str | None = None,
) -> bool:
    """Partial update of a subject row.

    Only kwargs with non-None values are written. ``created_at`` is
    intentionally NOT updatable here. Use raw SQL if you really need
    it.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        subject_id: Row to update.
        subject_code: New code (if not None).
        sex: New sex (if not None).
        origin: New origin (if not None).

    Returns:
        ``True`` iff exactly one row was updated. ``False`` (no SQL
        run) when every kwarg is None.
    """
    fields = {
        "subject_code": subject_code,
        "sex": sex,
        "origin": origin,
    }
    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(subject_id)
    cur.execute(
        f"UPDATE subjects SET {set_clause} WHERE subject_id = ?", tuple(params)
    )
    return cur.rowcount > 0

delete

delete(cur, subject_id: int) -> bool

Delete a subject.

visits.fk_visits_subject is ON DELETE CASCADE, so this also removes every visit, sample, and metadata row owned by the subject. sample_files uses ON DELETE RESTRICT and will block the delete instead — clean those up first.

Parameters:

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

Returns:

Type	Description
bool	True iff a row was removed.

Source code in src/noxdb/subjects.py

def delete(cur, subject_id: int) -> bool:
    """Delete a subject.

    ``visits.fk_visits_subject`` is ``ON DELETE CASCADE``, so this also
    removes every visit, sample, and metadata row owned by the subject.
    ``sample_files`` uses ``ON DELETE RESTRICT`` and will block the
    delete instead — clean those up first.

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

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

exists

exists(cur, subject_id: int) -> bool

Return whether a subject with the given id exists.

For natural-key existence checks use get_by_code(...) is not None.

Parameters:

Name	Type	Description	Default
cur		Audit-logging cursor from transaction().	required
subject_id	int	Id to check.	required

Returns:

Type	Description
bool	True if a matching row exists.

Source code in src/noxdb/subjects.py

def exists(cur, subject_id: int) -> bool:
    """Return whether a subject with the given id exists.

    For natural-key existence checks use
    ``get_by_code(...) is not None``.

    Args:
        cur: Audit-logging cursor from `transaction()`.
        subject_id: Id to check.

    Returns:
        ``True`` if a matching row exists.
    """
    cur.execute("SELECT 1 FROM subjects WHERE subject_id = ?", (subject_id,))
    return cur.fetchone() is not None