Database Reference

The database package is responsible for persistent state, schema migration, query execution, and import/seed workflows.

Package entrypoint

Purpose: export surface for DB helpers.

Database package — single authoritative SQLite path via schema.py + queries.py.

add_student(last_name, first_name, school='', grade='', preferred_formats='', notes='')

Create a student record and return the new id.

Source code in accessibility_mgr/db/queries.py

def add_student(
    last_name: str, first_name: str,
    school: str = "", grade: str = "",
    preferred_formats: str = "", notes: str = "",
) -> int:
    """Create a student record and return the new id."""
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO student (last_name,first_name,school,grade,preferred_formats,notes) "
            "VALUES (?,?,?,?,?,?)",
            (last_name, first_name, school, grade, preferred_formats, notes),
        )
        return int(cur.lastrowid)

backfill_metadata_keys(approved_keys)

Normalise and backfill typo'd metadata keys into approved keys (writes to DB).

Source code in accessibility_mgr/db/queries.py

def backfill_metadata_keys(approved_keys: list[str]) -> dict[str, Any]:
    """Normalise and backfill typo'd metadata keys into approved keys (writes to DB)."""
    if not approved_keys:
        return {"updated_rows": 0, "deleted_rows": 0, "mappings": {}, "skipped_keys": []}

    def _norm(k: str) -> str:
        k = k.strip().lower().replace(" ", "_").replace("-", "_")
        return re.sub(r"[^a-z0-9_:]", "", k)

    approved_set = set(approved_keys)
    norm_to_key = {_norm(k): k for k in approved_keys}
    norm_candidates = list(norm_to_key.keys())

    with get_conn() as conn:
        distinct = _rows(conn.execute("SELECT DISTINCT meta_key FROM job_metadata ORDER BY meta_key"))

        mappings: dict[str, str] = {}
        skipped: list[str] = []

        for row in distinct:
            source = row["meta_key"]
            if source in approved_set:
                continue
            nsrc = _norm(source)

            if nsrc in norm_to_key:
                mappings[source] = norm_to_key[nsrc]
                continue

            closest = difflib.get_close_matches(nsrc, norm_candidates, n=1, cutoff=0.8)
            if closest:
                mappings[source] = norm_to_key[closest[0]]
            else:
                skipped.append(source)

        updated_rows = 0
        deleted_rows = 0

        for source, target in mappings.items():
            if source == target:
                continue

            rows = _rows(conn.execute(
                "SELECT id, job_type, job_id, meta_value FROM job_metadata WHERE meta_key=?",
                (source,),
            ))

            for r in rows:
                existing = _rows(conn.execute(
                    "SELECT id, meta_value FROM job_metadata "
                    "WHERE job_type=? AND job_id=? AND meta_key=?",
                    (r["job_type"], r["job_id"], target),
                ))

                if not existing:
                    conn.execute(
                        "UPDATE job_metadata SET meta_key=?, updated_at=datetime('now') WHERE id=?",
                        (target, r["id"]),
                    )
                    updated_rows += 1
                    continue

                tgt_id = existing[0]["id"]
                tgt_val = existing[0].get("meta_value") or ""
                src_val = r.get("meta_value") or ""

                merged = tgt_val
                if src_val and src_val not in tgt_val:
                    merged = f"{tgt_val} | {src_val}" if tgt_val else src_val
                    conn.execute(
                        "UPDATE job_metadata SET meta_value=?, updated_at=datetime('now') WHERE id=?",
                        (merged, tgt_id),
                    )
                    updated_rows += 1

                conn.execute("DELETE FROM job_metadata WHERE id=?", (r["id"],))
                deleted_rows += 1

        return {
            "updated_rows": updated_rows,
            "deleted_rows": deleted_rows,
            "mappings": mappings,
            "skipped_keys": skipped,
        }

complete_step(job_type, job_id, step_key, agent='user')

Mark a workflow step as complete and log a STEP_COMPLETE event.

Source code in accessibility_mgr/db/queries.py

def complete_step(job_type: str, job_id: int, step_key: str, agent: str = "user") -> None:
    """Mark a workflow step as complete and log a STEP_COMPLETE event."""
    table = _STEP_TABLES.get(job_type)
    if not table or step_key not in _ALLOWED_STEPS.get(job_type, []):
        raise ValueError(f"Unknown step '{step_key}' for job type '{job_type}'")
    step_date_col = f"{step_key}_date"
    with get_conn() as conn:
        conn.execute(
            f"UPDATE {table} SET {step_key} = 1, {step_date_col} = datetime('now'), updated_at = datetime('now') WHERE id = ?",  # noqa: S608 - table/step/date columns come from fixed maps, never raw user SQL.
            (job_id,),
        )
    log_event(job_type, job_id, "STEP_COMPLETE", "SUCCESS",
              step_key=step_key, agent=agent, detail=f"Step '{step_key}' marked complete")

count_jobs_for_students(student_ids)

Return total job counts for a batch of students in a single query.

Issues one SQL UNION ALL query rather than four separate queries per student. Returns {student_id: total_count}.

Source code in accessibility_mgr/db/queries.py

def count_jobs_for_students(student_ids: list[int]) -> dict[int, int]:
    """Return total job counts for a batch of students in a single query.

    Issues one SQL UNION ALL query rather than four separate queries per
    student. Returns {student_id: total_count}.
    """
    if not student_ids:
        return {}
    placeholders = ",".join("?" * len(student_ids))
    with get_conn() as conn:
        rows = conn.execute(
            f"""
            SELECT student_id, COUNT(*) AS cnt FROM (
                SELECT student_id FROM braille_job         WHERE student_id IN ({placeholders})
                UNION ALL
                SELECT student_id FROM lp_ebraille_job     WHERE student_id IN ({placeholders})
                UNION ALL
                SELECT student_id FROM tactile_graphics_job WHERE student_id IN ({placeholders})
                UNION ALL
                SELECT student_id FROM print_job            WHERE student_id IN ({placeholders})
            ) GROUP BY student_id
            """,  # noqa: S608 - placeholders are '?' only; no user content in SQL.
            student_ids * 4,
        ).fetchall()
    return {row[0]: row[1] for row in rows}

deduct_filament(row_id, grams)

Deduct grams from filament stock.

FUN-011: grams must be strictly positive. Zero is a no-op; negative values would silently add stock (MAX(0, qty - negative) = qty + |negative|).

Source code in accessibility_mgr/db/queries.py

def deduct_filament(row_id: int, grams: float) -> None:
    """Deduct *grams* from filament stock.

    FUN-011: grams must be strictly positive.  Zero is a no-op; negative values
    would silently *add* stock (MAX(0, qty - negative) = qty + |negative|).
    """
    if grams <= 0:
        raise ValueError(f"grams must be positive, got {grams!r}")
    with get_conn() as conn:
        conn.execute(
            "UPDATE filament SET quantity_g = MAX(0, quantity_g - ?), "
            "updated_at = datetime('now') WHERE id = ?",
            (grams, row_id),
        )

delete_braille_job(row_id)

Delete a braille job, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_braille_job(row_id: int) -> None:
    """Delete a braille job, logging a DELETE audit event first (FIX-002)."""
    old = get_braille_job(row_id)
    if old:
        log_event(
            "braille", row_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"Job deleted: {old.get('title', '')}",
            extra_metadata={k: old.get(k) for k in
                            ["title", "braille_type", "requester", "priority", "created_at"]},
        )
    with get_conn() as conn:
        conn.execute("DELETE FROM braille_job WHERE id = ?", (row_id,))
        _delete_job_orphans(conn, "braille", row_id)

delete_file_object(file_id)

Delete a file object, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_file_object(file_id: int) -> None:
    """Delete a file object, logging a DELETE audit event first (FIX-002)."""
    row = get_file_object(file_id)
    if row:
        # FIX-002: log before deleting
        log_event(
            "file", file_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"File object deleted: {row.get('original_name', '')}",
            extra_metadata={k: row.get(k) for k in
                            ["original_name", "stored_path", "checksum_sha256", "file_use"]},
        )
        _sp = Path(row["stored_path"])
        stored = _sp if _sp.is_absolute() else FILES_DIR / _sp
        stored = stored.resolve()
        # SEC-004: only unlink if the resolved path is inside ARTIFACTS_DIR or FILES_DIR
        safe_roots = (ARTIFACTS_DIR.resolve(), FILES_DIR.resolve())
        if any(str(stored).startswith(str(root)) for root in safe_roots):
            stored.unlink(missing_ok=True)
        else:
            import logging as _log
            _log.getLogger(__name__).warning(
                "delete_file_object: refusing to unlink '%s' outside permitted dirs", stored
            )
    with get_conn() as conn:
        conn.execute("DELETE FROM file_object WHERE id = ?", (file_id,))

delete_lp_job(row_id)

Delete an LP/eBraille job, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_lp_job(row_id: int) -> None:
    """Delete an LP/eBraille job, logging a DELETE audit event first (FIX-002)."""
    old = get_lp_job(row_id)
    if old:
        log_event(
            "lp_ebraille", row_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"Job deleted: {old.get('title', '')}",
            extra_metadata={k: old.get(k) for k in
                            ["title", "job_type", "requester", "priority", "created_at"]},
        )
    with get_conn() as conn:
        conn.execute("DELETE FROM lp_ebraille_job WHERE id = ?", (row_id,))
        _delete_job_orphans(conn, "lp_ebraille", row_id)

delete_material_category(row_id)

Soft-delete a material category (sets active=0).

Source code in accessibility_mgr/db/queries.py

def delete_material_category(row_id: int) -> None:
    """Soft-delete a material category (sets active=0)."""
    set_material_category_active(row_id, 0)

delete_print_job(row_id)

Delete a print job, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_print_job(row_id: int) -> None:
    """Delete a print job, logging a DELETE audit event first (FIX-002)."""
    old = get_print_job(row_id)
    if old:
        log_event(
            "print", row_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"Print job deleted: {old.get('object_name') or old.get('file_name') or 'unnamed'}",
            extra_metadata={k: old.get(k) for k in
                            ["object_name", "printer_id", "filament_used_g", "successful", "printed_at"]},
        )
    with get_conn() as conn:
        conn.execute("DELETE FROM print_job WHERE id = ?", (row_id,))
        _delete_job_orphans(conn, "print", row_id)

delete_student(student_id)

Soft-delete a student (sets active=0).

Source code in accessibility_mgr/db/queries.py

def delete_student(student_id: int) -> None:
    """Soft-delete a student (sets active=0)."""
    update_student(student_id, active=0)

delete_tactile_job(row_id)

Delete a tactile job, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_tactile_job(row_id: int) -> None:
    """Delete a tactile job, logging a DELETE audit event first (FIX-002)."""
    old = get_tactile_job(row_id)
    if old:
        log_event(
            "tactile", row_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"Job deleted: {old.get('title', '')}",
            extra_metadata={k: old.get(k) for k in
                            ["title", "tactile_type", "requester", "priority", "created_at"]},
        )
    with get_conn() as conn:
        conn.execute("DELETE FROM tactile_graphics_job WHERE id = ?", (row_id,))
        _delete_job_orphans(conn, "tactile", row_id)

delete_workflow_step(row_id)

Soft-delete a workflow step (sets active=0).

Source code in accessibility_mgr/db/queries.py

def delete_workflow_step(row_id: int) -> None:
    """Soft-delete a workflow step (sets active=0)."""
    set_workflow_step_active(row_id, 0)

get_print_job(row_id)

Fetch a single print job by id (FIX-001: needed for pre-update snapshot).

Source code in accessibility_mgr/db/queries.py

def get_print_job(row_id: int) -> Optional[dict[str, Any]]:
    """Fetch a single print job by id (FIX-001: needed for pre-update snapshot)."""
    with get_conn() as conn:
        rows = _rows(conn.execute("""
            SELECT pj.*,
                   p.name  AS printer_name,
                   f.brand || ' ' || f.color || ' ' || f.filament_type AS filament_desc
            FROM print_job pj
            LEFT JOIN printer  p ON p.id = pj.printer_id
            LEFT JOIN filament f ON f.id = pj.filament_id
            WHERE pj.id = ?
        """, (row_id,)))
        return rows[0] if rows else None

get_student(student_id)

Fetch a single student record by id.

Source code in accessibility_mgr/db/queries.py

def get_student(student_id: int) -> Optional[dict[str, Any]]:
    """Fetch a single student record by id."""
    with get_conn() as conn:
        rows = _rows(conn.execute("SELECT * FROM student WHERE id = ?", (student_id,)))
        return rows[0] if rows else None

ingest_file(source_path, file_use='ORIGINAL', format_name='', format_version='', encoding='', extra_metadata=None, project_title='', student_initials='', school_name='', grade_level='', subject='')

Copy a file into the artifact store, compute SHA-256, and create a file_object record.

SEC-004 / FUN-023: the source_path must exist and must resolve to a path inside FILES_DIR (the staging area). This prevents callers from staging arbitrary files from anywhere on the filesystem into the artifact store.

Source code in accessibility_mgr/db/queries.py

def ingest_file(
    source_path: str,
    file_use: str = "ORIGINAL",
    format_name: str = "",
    format_version: str = "",
    encoding: str = "",
    extra_metadata: Optional[dict[str, Any]] = None,
    project_title: str = "",
    student_initials: str = "",
    school_name: str = "",
    grade_level: str = "",
    subject: str = "",
) -> int:
    """Copy a file into the artifact store, compute SHA-256, and create a file_object record.

    SEC-004 / FUN-023: the *source_path* must exist and must resolve to a path
    inside FILES_DIR (the staging area).  This prevents callers from staging
    arbitrary files from anywhere on the filesystem into the artifact store.
    """
    src = Path(source_path).resolve()
    if not src.exists():
        raise FileNotFoundError(f"Source file not found: {src}")

    # SEC-004: ensure source resolves inside the permitted staging directory
    try:
        src.relative_to(FILES_DIR.resolve())
    except ValueError:
        raise PermissionError(
            f"ingest_file: source '{src}' is outside the permitted staging directory "
            f"'{FILES_DIR}'.  Stage files via the upload handler before ingesting."
        )

    file_uuid = str(uuid.uuid4())

    if project_title:
        safe_project_title = _sanitize_name(project_title) or file_uuid[:8]
        project_dir = ARTIFACTS_DIR / safe_project_title
        project_dir.mkdir(parents=True, exist_ok=True)

        name_parts: list[str] = []
        if student_initials:
            cleaned = _sanitize_name(student_initials)
            if cleaned:
                name_parts.append(cleaned)
        if school_name:
            cleaned = _sanitize_name(school_name)
            if cleaned:
                name_parts.append(cleaned)
        if grade_level:
            cleaned = _sanitize_name(grade_level)
            if cleaned:
                name_parts.append(f"Grade{cleaned}")
        if subject:
            cleaned = _sanitize_name(subject)
            if cleaned:
                name_parts.append(cleaned)

        artifact_stem = "_".join(name_parts) or file_uuid[:8]
        max_path_len = 240

        # Keep destination paths safely below typical filesystem path limits.
        while len(str(project_dir / f"{artifact_stem}{src.suffix}")) > max_path_len and len(artifact_stem) > 16:
            artifact_stem = artifact_stem[:-8]

        if len(f"{artifact_stem}{src.suffix}") > 255:
            raise ValueError("Artifact file name exceeds filesystem component limits")

        dest = project_dir / f"{artifact_stem}{src.suffix}"
        if len(str(dest)) > max_path_len:
            raise ValueError(
                "Artifact destination path is too long; shorten project or metadata values"
            )
        if dest.exists():
            dest = project_dir / f"{artifact_stem}_{file_uuid[:8]}{src.suffix}"
            if len(str(dest)) > max_path_len:
                raise ValueError(
                    "Artifact destination path is too long after collision handling"
                )
        stored_path_val = str(dest)
    else:
        dest = FILES_DIR / f"{file_uuid}{src.suffix}"
        stored_path_val = dest.name

    shutil.copy2(src, dest)
    checksum = _sha256(dest)
    size_bytes = dest.stat().st_size
    mime_type = mimetypes.guess_type(src.name)[0] or "application/octet-stream"

    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO file_object (uuid,original_name,stored_path,mime_type,size_bytes,"
            "checksum_sha256,file_use,format_name,format_version,encoding,extra_metadata) "
            "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
            (file_uuid, src.name, stored_path_val, mime_type, size_bytes, checksum,
             file_use, format_name, format_version, encoding,
             json.dumps(extra_metadata) if extra_metadata else None),
        )
        return int(cur.lastrowid)

list_distinct_metadata_keys()

Return all metadata keys in use with occurrence counts.

Source code in accessibility_mgr/db/queries.py

def list_distinct_metadata_keys() -> list[dict[str, Any]]:
    """Return all metadata keys in use with occurrence counts."""
    with get_conn() as conn:
        return _rows(conn.execute(
            "SELECT meta_key, COUNT(*) AS usage_count "
            "FROM job_metadata GROUP BY meta_key ORDER BY usage_count DESC, meta_key"
        ))

list_filaments()

Return all filament records ordered by brand and colour.

Source code in accessibility_mgr/db/queries.py

def list_filaments() -> list[dict[str, Any]]:
    """Return all filament records ordered by brand and colour."""
    with get_conn() as conn:
        return _rows(conn.execute("SELECT * FROM filament ORDER BY brand, color"))

list_jobs_for_student(student_id)

Return all jobs linked to a student, grouped by type.

Source code in accessibility_mgr/db/queries.py

def list_jobs_for_student(student_id: int) -> dict[str, list[dict[str, Any]]]:
    """Return all jobs linked to a student, grouped by type."""
    with get_conn() as conn:
        braille = _rows(conn.execute(
            "SELECT *, 'braille' AS job_type FROM braille_job WHERE student_id = ? ORDER BY created_at DESC",
            (student_id,),
        ))
        lp = _rows(conn.execute(
            "SELECT *, job_type FROM lp_ebraille_job WHERE student_id = ? ORDER BY created_at DESC",
            (student_id,),
        ))
        tactile = _rows(conn.execute(
            "SELECT *, 'tactile' AS job_type FROM tactile_graphics_job WHERE student_id = ? ORDER BY created_at DESC",
            (student_id,),
        ))
        print_jobs = _rows(conn.execute(
            "SELECT *, 'print' AS job_type FROM print_job WHERE student_id = ? ORDER BY printed_at DESC",
            (student_id,),
        ))
    return {
        "braille": braille,
        "lp_ebraille": lp,
        "tactile": tactile,
        "print": print_jobs,
    }

list_students(active_only=True)

Return student records ordered by last name, first name.

Source code in accessibility_mgr/db/queries.py

def list_students(active_only: bool = True) -> list[dict[str, Any]]:
    """Return student records ordered by last name, first name."""
    with get_conn() as conn:
        where = "WHERE active = 1" if active_only else ""
        return _rows(conn.execute(
            f"SELECT * FROM student {where} ORDER BY last_name, first_name"  # noqa: S608 - where clause is a fixed toggle, no user-supplied SQL.
        ))

list_students_page(search=None, active_only=True, limit=50, offset=0)

Return a paginated page of student records with optional search.

Source code in accessibility_mgr/db/queries.py

def list_students_page(
    search: Optional[str] = None,
    active_only: bool = True,
    limit: int = 50,
    offset: int = 0,
) -> list[dict[str, Any]]:
    """Return a paginated page of student records with optional search."""
    filters: list[str] = []
    params: list[Any] = []
    if active_only:
        filters.append("active = 1")
    if search:
        term = f"%{search}%"
        filters.append("(last_name LIKE ? OR first_name LIKE ? OR school LIKE ?)")
        params.extend([term, term, term])
    where = ("WHERE " + " AND ".join(filters)) if filters else ""
    params.extend([limit, max(0, offset)])
    with get_conn() as conn:
        return _rows(conn.execute(
            f"SELECT * FROM student {where} ORDER BY last_name, first_name LIMIT ? OFFSET ?",  # noqa: S608 - where clause from fixed SQL fragments; values parameterised.
            params,
        ))

log_qa_measure(*, engine, epub_path, passed, score, error_count=0, warning_count=0, info_count=0, issues=None, job_type=None, job_id=None, reviewer=None, reviewer_notes='', checked_at=None)

Persist a reviewer-submitted QA measure and return its row id.

When the measure is linked to a job (job_type + job_id), this also writes a QA_MEASURE_SUBMITTED entry to that job's event log so the result is visible alongside the job's other history, consistent with how log_qa_run results are linked via FIX-012 in ui/qa.py.

Source code in accessibility_mgr/db/queries.py

def log_qa_measure(
    *,
    engine: str,
    epub_path: str,
    passed: bool,
    score: int,
    error_count: int = 0,
    warning_count: int = 0,
    info_count: int = 0,
    issues: Optional[list[dict[str, Any]]] = None,
    job_type: Optional[str] = None,
    job_id: Optional[int] = None,
    reviewer: Optional[str] = None,
    reviewer_notes: str = "",
    checked_at: Optional[str] = None,
) -> int:
    """Persist a reviewer-submitted QA measure and return its row id.

    When the measure is linked to a job (job_type + job_id), this also
    writes a QA_MEASURE_SUBMITTED entry to that job's event log so the
    result is visible alongside the job's other history, consistent with
    how log_qa_run results are linked via FIX-012 in ui/qa.py.
    """
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO qa_measure (engine,epub_path,job_type,job_id,passed,"
            "score,error_count,warning_count,info_count,issues,reviewer,"
            "reviewer_notes,checked_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                engine, epub_path, job_type, job_id, 1 if passed else 0, score,
                error_count, warning_count, info_count,
                json.dumps(issues or []), reviewer, reviewer_notes, checked_at,
            ),
        )
        measure_id = int(cur.lastrowid)

    if job_type and job_id:
        log_event(
            job_type, job_id, "QA_MEASURE_SUBMITTED",
            event_outcome="SUCCESS" if passed else "FAILURE",
            agent=reviewer or "system",
            detail=(
                f"{engine} accessibility QA — score {score}, "
                f"{error_count} error(s), {warning_count} warning(s)"
            ),
            extra_metadata={"qa_measure_id": measure_id, "score": score},
        )

    return measure_id

preview_backfill_metadata_keys(approved_keys)

Return proposed key mappings without writing to the database (FIX-017 dry-run).

Source code in accessibility_mgr/db/queries.py

def preview_backfill_metadata_keys(approved_keys: list[str]) -> dict[str, Any]:
    """Return proposed key mappings without writing to the database (FIX-017 dry-run)."""
    if not approved_keys:
        return {"mappings": {}, "skipped_keys": [], "usage_counts": {}}

    def _norm(k: str) -> str:
        k = k.strip().lower().replace(" ", "_").replace("-", "_")
        return re.sub(r"[^a-z0-9_:]", "", k)

    approved_set = set(approved_keys)
    norm_to_key = {_norm(k): k for k in approved_keys}
    norm_candidates = list(norm_to_key.keys())

    with get_conn() as conn:
        distinct = _rows(conn.execute(
            "SELECT meta_key, COUNT(*) AS usage_count FROM job_metadata "
            "GROUP BY meta_key ORDER BY meta_key"
        ))

    mappings: dict[str, str] = {}
    skipped: list[str] = []
    usage_counts: dict[str, int] = {r["meta_key"]: r["usage_count"] for r in distinct}

    for row in distinct:
        source = row["meta_key"]
        if source in approved_set:
            continue
        nsrc = _norm(source)

        if nsrc in norm_to_key:
            mappings[source] = norm_to_key[nsrc]
            continue

        closest = difflib.get_close_matches(nsrc, norm_candidates, n=1, cutoff=0.8)
        if closest:
            mappings[source] = norm_to_key[closest[0]]
        else:
            skipped.append(source)

    return {
        "mappings": mappings,
        "skipped_keys": skipped,
        "usage_counts": usage_counts,
    }

record_delivery(job_type, job_id, delivery_method, delivery_recipient, delivery_date=None, delivery_notes='', agent='user')

Record delivery details, complete the 'delivered' step, and log a DELIVERY event (FIX-016).

Source code in accessibility_mgr/db/queries.py

def record_delivery(
    job_type: str, job_id: int,
    delivery_method: str,
    delivery_recipient: str,
    delivery_date: Optional[str] = None,
    delivery_notes: str = "",
    agent: str = "user",
) -> None:
    """Record delivery details, complete the 'delivered' step, and log a DELIVERY event (FIX-016)."""
    from datetime import date
    d_date = delivery_date or date.today().isoformat()

    # Update delivery columns directly without emitting a redundant FIELD_UPDATE event.
    _table_map = {
        "braille":     "braille_job",
        "lp_ebraille": "lp_ebraille_job",
        "tactile":     "tactile_graphics_job",
        "print":       "print_job",
    }
    table = _table_map.get(job_type)
    if table and table in _SAFE_TABLES:
        with get_conn() as conn:
            conn.execute(  # noqa: S608 - table comes from a fixed allow-map, not user input.
                f"UPDATE {table} SET delivered=1, delivery_date=?, delivery_method=?, "
                f"delivery_recipient=?, delivery_notes=?, updated_at=datetime('now') WHERE id=?",
                (d_date, delivery_method, delivery_recipient, delivery_notes, job_id),
            )

    log_event(
        job_type, job_id, "DELIVERY", "SUCCESS",
        step_key="delivered",
        agent=agent,
        detail=f"Delivered to {delivery_recipient} via {delivery_method} on {d_date}",
        extra_metadata={
            "delivery_method": delivery_method,
            "delivery_recipient": delivery_recipient,
            "delivery_date": d_date,
            "delivery_notes": delivery_notes,
        },
    )

report_jobs(school=None, grade=None, job_type=None, status=None, priority=None, date_from=None, date_to=None, student_id=None)

Return filtered job lists grouped by type with summary counts.

Parameters

school : str, optional Filter by student.school (exact) or dc:coverage metadata value (LIKE). grade : str, optional Filter by student.grade or grade_level metadata value (LIKE). job_type : str, optional One of 'braille', 'lp_ebraille', 'tactile', 'print'. If None, all types returned. status : str, optional 'not_started', 'in_progress', or 'delivered'. priority : str, optional One of 'low', 'normal', 'high', 'urgent'. date_from : str, optional ISO date string — jobs created on or after this date. date_to : str, optional ISO date string — jobs created on or before this date. student_id : int, optional Return only jobs for this student.

Source code in accessibility_mgr/db/queries.py

def report_jobs(
    school: Optional[str] = None,
    grade: Optional[str] = None,
    job_type: Optional[str] = None,
    status: Optional[str] = None,
    priority: Optional[str] = None,
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
    student_id: Optional[int] = None,
) -> dict[str, Any]:
    """Return filtered job lists grouped by type with summary counts.

    Parameters
    ----------
    school : str, optional
        Filter by student.school (exact) or dc:coverage metadata value (LIKE).
    grade : str, optional
        Filter by student.grade or grade_level metadata value (LIKE).
    job_type : str, optional
        One of 'braille', 'lp_ebraille', 'tactile', 'print'. If None, all types returned.
    status : str, optional
        'not_started', 'in_progress', or 'delivered'.
    priority : str, optional
        One of 'low', 'normal', 'high', 'urgent'.
    date_from : str, optional
        ISO date string — jobs created on or after this date.
    date_to : str, optional
        ISO date string — jobs created on or before this date.
    student_id : int, optional
        Return only jobs for this student.
    """

    def _step_status_expr(steps: list[str]) -> str:
        """Build SQL CASE expression deriving a status label from step columns."""
        delivered_col = steps[-1]
        any_done = " + ".join(f"COALESCE({s}, 0)" for s in steps)
        return (
            f"CASE WHEN {delivered_col} = 1 THEN 'delivered' "
            f"WHEN ({any_done}) > 0 THEN 'in_progress' "
            f"ELSE 'not_started' END"
        )

    def _run(
        table: str,
        type_label: str,
        steps: list[str],
        title_col: str = "title",
        date_col: str = "created_at",
        priority_expr: str = "j.priority",
        type_expr: str | None = None,
    ) -> list[dict[str, Any]]:
        # SAFE: all SQL-interpolated identifiers come from fixed in-function constants,
        # not user input. User-supplied values are bound via '?' parameterisation only.
        filters: list[str] = []
        params: list[Any] = []

        if student_id is not None:
            filters.append("j.student_id = ?")
            params.append(student_id)
        if date_from:
            filters.append(f"j.{date_col} >= ?")
            params.append(date_from)
        if date_to:
            filters.append(f"j.{date_col} <= ?")
            params.append(date_to)
        if school:
            filters.append(
                "(s.school LIKE ? OR EXISTS ("
                "  SELECT 1 FROM job_metadata jm "
                "  WHERE jm.job_type=? AND jm.job_id=j.id "
                "  AND jm.meta_key='dc:coverage' AND jm.meta_value LIKE ?"
                "))"
            )
            params += [f"%{school}%", type_label, f"%{school}%"]
        if grade:
            filters.append(
                "(s.grade LIKE ? OR EXISTS ("
                "  SELECT 1 FROM job_metadata jm "
                "  WHERE jm.job_type=? AND jm.job_id=j.id "
                "  AND jm.meta_key='grade_level' AND jm.meta_value LIKE ?"
                "))"
            )
            params += [f"%{grade}%", type_label, f"%{grade}%"]

        status_expr = _step_status_expr(steps)
        if status:
            filters.append(f"({status_expr}) = ?")
            params.append(status)
        if priority and priority_expr != "NULL":
            filters.append(f"{priority_expr} = ?")
            params.append(priority)

        where = ("WHERE " + " AND ".join(filters)) if filters else ""
        resolved_type_expr = type_expr or f"'{type_label}'"

        sql = f"""
                                 SELECT j.id, j.{title_col} AS title, {resolved_type_expr} AS job_type,
                 j.requester, {priority_expr} AS priority, j.{date_col} AS created_at,
                   s.last_name, s.first_name, s.school, s.grade,
                   ({status_expr}) AS status
            FROM {table} j
            LEFT JOIN student s ON s.id = j.student_id
            {where}
                 ORDER BY j.{date_col} DESC
        """  # noqa: S608 - table/column identifiers come from trusted in-module constants.
        with get_conn() as conn:
            return _rows(conn.execute(sql, params))

    b_steps   = ["digitized", "formatted", "brailled", "proofread", "delivered"]
    lp_steps  = ["digitized", "formatted", "converted", "proofread", "delivered"]
    tac_steps = ["designed", "produced", "qa_reviewed", "delivered"]
    prt_steps = ["designed", "sliced", "printed", "inspected", "delivered"]

    results: dict[str, list[dict[str, Any]]] = {}

    if job_type in (None, "braille"):
        results["braille"] = _run("braille_job", "braille", b_steps)
    if job_type in (None, "lp_ebraille"):
        results["lp_ebraille"] = _run(
            "lp_ebraille_job",
            "lp_ebraille",
            lp_steps,
            type_expr="COALESCE(j.job_type, 'lp_ebraille')",
        )
    if job_type in (None, "tactile"):
        results["tactile"] = _run("tactile_graphics_job", "tactile", tac_steps)
    if job_type in (None, "print"):
        results["print"] = _run(
            "print_job",
            "print",
            prt_steps,
            title_col="object_name",
            date_col="printed_at",
            priority_expr="'normal'",
        )

    all_jobs: list[dict[str, Any]] = []
    for rows in results.values():
        all_jobs.extend(rows)

    by_type = {k: len(v) for k, v in results.items()}
    return {
        "total_jobs": len(all_jobs),
        "by_type": by_type,
        "jobs": all_jobs,
        "by_type_lists": results,
    }

revert_step(job_type, job_id, step_key, agent='user', reason='')

Revert a workflow step and log a STEP_REVERT event.

Source code in accessibility_mgr/db/queries.py

def revert_step(
    job_type: str, job_id: int, step_key: str,
    agent: str = "user", reason: str = "",
) -> None:
    """Revert a workflow step and log a STEP_REVERT event."""
    table = _STEP_TABLES.get(job_type)
    if not table or step_key not in _ALLOWED_STEPS.get(job_type, []):
        raise ValueError(f"Unknown step '{step_key}' for job type '{job_type}'")
    step_date_col = f"{step_key}_date"
    with get_conn() as conn:
        conn.execute(
            f"UPDATE {table} SET {step_key} = 0, {step_date_col} = NULL, updated_at = datetime('now') WHERE id = ?",  # noqa: S608 - table/step/date columns come from fixed maps, never raw user SQL.
            (job_id,),
        )
    log_event(job_type, job_id, "STEP_REVERT", "WARNING",
              step_key=step_key, agent=agent,
              detail=f"Step '{step_key}' reverted" + (f": {reason}" if reason else ""))

search_all(query, limit=200)

Search all job tables, files, metadata, and event log using SQL LIKE queries.

Replaces the in-memory Python filtering in the UI layer (FIX-009). Also searches event log detail text (FIX-014) and file checksums.

Parameters

query : str The search term. Applied with LIKE '%term%' across text columns. If exactly 64 hex characters, also tested as an exact SHA-256 match. limit : int Maximum rows returned per result category.

Source code in accessibility_mgr/db/queries.py

def search_all(query: str, limit: int = 200) -> dict[str, list[dict[str, Any]]]:
    """Search all job tables, files, metadata, and event log using SQL LIKE queries.

    Replaces the in-memory Python filtering in the UI layer (FIX-009).
    Also searches event log detail text (FIX-014) and file checksums.

    Parameters
    ----------
    query : str
        The search term. Applied with LIKE '%term%' across text columns.
        If exactly 64 hex characters, also tested as an exact SHA-256 match.
    limit : int
        Maximum rows returned per result category.
    """
    term = f"%{query}%"
    # SHA-256 exact match for 64-char hex strings
    is_checksum = len(query) == 64 and all(c in "0123456789abcdefABCDEF" for c in query)

    with get_conn() as conn:
        if not is_checksum:
            try:
                def _fts_ids(table: str) -> list[int]:
                    rows = conn.execute(
                        f"SELECT id FROM {table} WHERE {table} MATCH ? LIMIT ?",  # noqa: S608 - table comes from fixed in-function constants.
                        (query, limit),
                    ).fetchall()
                    return [int(r[0]) for r in rows]

                def _rows_by_ids(
                    table: str,
                    columns: str,
                    order_by: str,
                    ids: list[int],
                ) -> list[dict[str, Any]]:
                    if not ids:
                        return []
                    placeholders = ",".join("?" for _ in ids)
                    sql = (
                        f"SELECT {columns} FROM {table} "
                        f"WHERE id IN ({placeholders}) ORDER BY {order_by} LIMIT ?"
                    )
                    return _rows(conn.execute(sql, [*ids, limit]))

                braille_jobs = _rows_by_ids(
                    "braille_job",
                    "id, title, braille_type, requester, priority, created_at",
                    "created_at DESC",
                    _fts_ids("braille_job_fts"),
                )
                lp_jobs = _rows_by_ids(
                    "lp_ebraille_job",
                    "id, title, job_type, requester, priority, created_at",
                    "created_at DESC",
                    _fts_ids("lp_ebraille_job_fts"),
                )
                tactile_jobs = _rows_by_ids(
                    "tactile_graphics_job",
                    "id, title, tactile_type, requester, priority, created_at",
                    "created_at DESC",
                    _fts_ids("tactile_graphics_job_fts"),
                )
                print_jobs = _rows_by_ids(
                    "print_job",
                    "id, object_name, file_name, requester, successful, printed_at",
                    "printed_at DESC",
                    _fts_ids("print_job_fts"),
                )
                files = _rows_by_ids(
                    "file_object",
                    "id, original_name, stored_path, file_use, format_name, checksum_sha256, created_at",
                    "created_at DESC",
                    _fts_ids("file_object_fts"),
                )
                metadata = _rows(conn.execute(
                    "SELECT job_type, CAST(job_id AS INTEGER) AS job_id, meta_key, meta_value "
                    "FROM job_metadata_fts WHERE job_metadata_fts MATCH ? LIMIT ?",
                    (query, limit),
                ))
                events = _rows(conn.execute(
                    "SELECT id, job_type, CAST(job_id AS INTEGER) AS job_id, event_type, agent, detail "
                    "FROM metadata_event_fts WHERE metadata_event_fts MATCH ? LIMIT ?",
                    (query, limit),
                ))
                students = _rows(conn.execute(
                    "SELECT id, last_name, first_name, school, grade "
                    "FROM student WHERE last_name LIKE ? OR first_name LIKE ? "
                    "OR school LIKE ? OR notes LIKE ? ORDER BY last_name LIMIT ?",
                    (term, term, term, term, limit),
                ))

                return {
                    "braille_jobs": braille_jobs,
                    "lp_jobs": lp_jobs,
                    "tactile_jobs": tactile_jobs,
                    "print_jobs": print_jobs,
                    "files": files,
                    "metadata": metadata,
                    "events": events,
                    "students": students,
                }
            except sqlite3.OperationalError:
                # Fallback for environments where FTS tables are unavailable.
                pass

        braille_jobs = _rows(conn.execute(
            "SELECT id, title, braille_type, requester, priority, created_at "
            "FROM braille_job WHERE title LIKE ? OR requester LIKE ? OR notes LIKE ? "
            "ORDER BY created_at DESC LIMIT ?",
            (term, term, term, limit),
        ))

        lp_jobs = _rows(conn.execute(
            "SELECT id, title, job_type, requester, priority, created_at "
            "FROM lp_ebraille_job WHERE title LIKE ? OR requester LIKE ? OR notes LIKE ? "
            "ORDER BY created_at DESC LIMIT ?",
            (term, term, term, limit),
        ))

        tactile_jobs = _rows(conn.execute(
            "SELECT id, title, tactile_type, requester, priority, created_at "
            "FROM tactile_graphics_job WHERE title LIKE ? OR requester LIKE ? OR notes LIKE ? "
            "ORDER BY created_at DESC LIMIT ?",
            (term, term, term, limit),
        ))

        print_jobs = _rows(conn.execute(
            "SELECT id, object_name, file_name, requester, successful, printed_at "
            "FROM print_job WHERE object_name LIKE ? OR requester LIKE ? "
            "OR file_name LIKE ? OR notes LIKE ? ORDER BY printed_at DESC LIMIT ?",
            (term, term, term, term, limit),
        ))

        if is_checksum:
            files = _rows(conn.execute(
                "SELECT id, original_name, stored_path, file_use, format_name, "
                "checksum_sha256, created_at FROM file_object "
                "WHERE original_name LIKE ? OR stored_path LIKE ? OR format_name LIKE ? "
                "OR encoding LIKE ? OR checksum_sha256 = ? ORDER BY created_at DESC LIMIT ?",
                (term, term, term, term, query, limit),
            ))
        else:
            files = _rows(conn.execute(
                "SELECT id, original_name, stored_path, file_use, format_name, "
                "checksum_sha256, created_at FROM file_object "
                "WHERE original_name LIKE ? OR stored_path LIKE ? OR format_name LIKE ? "
                "OR encoding LIKE ? ORDER BY created_at DESC LIMIT ?",
                (term, term, term, term, limit),
            ))

        metadata = _rows(conn.execute(
            "SELECT jm.job_type, jm.job_id, jm.meta_key, jm.meta_value "
            "FROM job_metadata jm WHERE jm.meta_key LIKE ? OR jm.meta_value LIKE ? "
            "ORDER BY jm.job_type, jm.job_id LIMIT ?",
            (term, term, limit),
        ))

        # FIX-014: event log search
        events = _rows(conn.execute(
            "SELECT me.id, me.job_type, me.job_id, me.event_type, me.agent, "
            "me.detail, me.event_datetime "
            "FROM metadata_event me "
            "WHERE me.detail LIKE ? OR me.agent LIKE ? OR me.event_type LIKE ? "
            "ORDER BY me.event_datetime DESC LIMIT ?",
            (term, term, term, limit),
        ))

        students = _rows(conn.execute(
            "SELECT id, last_name, first_name, school, grade "
            "FROM student WHERE last_name LIKE ? OR first_name LIKE ? "
            "OR school LIKE ? OR notes LIKE ? ORDER BY last_name LIMIT ?",
            (term, term, term, term, limit),
        ))

    return {
        "braille_jobs": braille_jobs,
        "lp_jobs": lp_jobs,
        "tactile_jobs": tactile_jobs,
        "print_jobs": print_jobs,
        "files": files,
        "metadata": metadata,
        "events": events,
        "students": students,
    }

update_braille_job(row_id, **fields)

Update a braille job and log a FIELD_UPDATE audit event (FIX-001, FIX-016).

Source code in accessibility_mgr/db/queries.py

def update_braille_job(row_id: int, **fields: Any) -> None:
    """Update a braille job and log a FIELD_UPDATE audit event (FIX-001, FIX-016)."""
    old = get_braille_job(row_id)
    allowed = {
        "title", "braille_type", "embosser_id", "requester", "request_date", "due_date",
        "priority", "digitized", "formatted", "brailled", "proofread", "delivered",
        "notes", "student_id",
        # FIX-016: delivery columns
        "delivery_date", "delivery_method", "delivery_recipient", "delivery_notes",
    }
    sql, vals = _build_update_sql("braille_job", fields, allowed)
    with get_conn() as conn:
        conn.execute(sql, vals + [row_id])
    log_event(
        "braille", row_id, "FIELD_UPDATE", "SUCCESS",
        agent="system",
        detail=f"Updated fields: {list(fields.keys())}",
        extra_metadata={
            "changed_fields": list(fields.keys()),
            "previous_values": {k: old.get(k) for k in fields if old},
        },
    )

update_lp_job(row_id, **fields)

Update an LP/eBraille job and log a FIELD_UPDATE audit event (FIX-001, FIX-016).

Source code in accessibility_mgr/db/queries.py

def update_lp_job(row_id: int, **fields: Any) -> None:
    """Update an LP/eBraille job and log a FIELD_UPDATE audit event (FIX-001, FIX-016)."""
    old = get_lp_job(row_id)
    allowed = {
        "title", "job_type", "requester", "request_date", "due_date",
        "priority", "digitized", "formatted", "converted", "proofread",
        "delivered", "notes", "student_id",
        # FIX-016
        "delivery_date", "delivery_method", "delivery_recipient", "delivery_notes",
    }
    sql, vals = _build_update_sql("lp_ebraille_job", fields, allowed)
    with get_conn() as conn:
        conn.execute(sql, vals + [row_id])
    log_event(
        "lp_ebraille", row_id, "FIELD_UPDATE", "SUCCESS",
        agent="system",
        detail=f"Updated fields: {list(fields.keys())}",
        extra_metadata={
            "changed_fields": list(fields.keys()),
            "previous_values": {k: old.get(k) for k in fields if old},
        },
    )

update_print_job(row_id, **fields)

Update a print job and log a FIELD_UPDATE audit event (FIX-001, FIX-016).

Source code in accessibility_mgr/db/queries.py

def update_print_job(row_id: int, **fields: Any) -> None:
    """Update a print job and log a FIELD_UPDATE audit event (FIX-001, FIX-016)."""
    # FIX-001: snapshot before update
    old = get_print_job(row_id)
    allowed = {
        "printer_id", "filament_id", "filament_used_g", "successful",
        "failure_reason", "object_name", "requester", "request_date", "notes",
        "student_id",
        # FIX-007: step columns
        "designed", "sliced", "printed", "inspected", "delivered",
        # FIX-016: delivery columns
        "delivery_date", "delivery_method", "delivery_recipient", "delivery_notes",
    }
    sql, vals = _build_update_sql("print_job", fields, allowed, has_updated_at=True)
    with get_conn() as conn:
        conn.execute(sql, vals + [row_id])
    # FIX-001: log what changed
    log_event(
        "print", row_id, "FIELD_UPDATE", "SUCCESS",
        agent="system",
        detail=f"Updated fields: {list(fields.keys())}",
        extra_metadata={
            "changed_fields": list(fields.keys()),
            "previous_values": {k: old.get(k) for k in fields if old},
        },
    )

update_student(student_id, **fields)

Update a student record.

Source code in accessibility_mgr/db/queries.py

def update_student(student_id: int, **fields: Any) -> None:
    """Update a student record."""
    allowed = {"last_name", "first_name", "school", "grade", "preferred_formats",
               "notes", "active"}
    sql, vals = _build_update_sql("student", fields, allowed)
    with get_conn() as conn:
        conn.execute(sql, vals + [student_id])

update_tactile_job(row_id, **fields)

Update a tactile job and log a FIELD_UPDATE audit event (FIX-001, FIX-016).

Source code in accessibility_mgr/db/queries.py

def update_tactile_job(row_id: int, **fields: Any) -> None:
    """Update a tactile job and log a FIELD_UPDATE audit event (FIX-001, FIX-016)."""
    old = get_tactile_job(row_id)
    allowed = {
        "title", "tactile_type", "requester", "request_date", "due_date",
        "priority", "designed", "produced", "qa_reviewed", "delivered",
        "notes", "student_id",
        # FIX-016
        "delivery_date", "delivery_method", "delivery_recipient", "delivery_notes",
    }
    sql, vals = _build_update_sql("tactile_graphics_job", fields, allowed)
    with get_conn() as conn:
        conn.execute(sql, vals + [row_id])
    log_event(
        "tactile", row_id, "FIELD_UPDATE", "SUCCESS",
        agent="system",
        detail=f"Updated fields: {list(fields.keys())}",
        extra_metadata={
            "changed_fields": list(fields.keys()),
            "previous_values": {k: old.get(k) for k in fields if old},
        },
    )

Schema and migration

Purpose: connection setup, path resolution, schema initialization, and incremental migration application.

Database schema, connection helpers, and initialisation.

Changes applied (see fix_specs.json): FIX-007 print_job gains five workflow step columns. FIX-010 student table added; all job tables gain student_id FK. FIX-011 file_use seed value changed from MASTER → ORIGINAL. FIX-016 All job tables gain delivery_date/method/recipient/notes columns.

get_conn()

Yield a configured SQLite connection; commit on success, rollback on error.

Source code in accessibility_mgr/db/schema.py

@contextmanager
def get_conn() -> Generator[sqlite3.Connection, None, None]:
    """Yield a configured SQLite connection; commit on success, rollback on error."""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("PRAGMA journal_mode = WAL")
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

init_db()

Create all tables, directories, and seed data if they do not exist.

Source code in accessibility_mgr/db/schema.py

def init_db() -> None:
    """Create all tables, directories, and seed data if they do not exist."""
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    PRINTS_DIR.mkdir(exist_ok=True)
    FILES_DIR.mkdir(exist_ok=True)
    BACKUPS_DIR.mkdir(exist_ok=True)
    ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True)
    with get_conn() as conn:
        conn.executescript(_SCHEMA_SQL)
        _migrate(conn)
    _validate_step_columns()

Query layer

Purpose: typed CRUD and workflow queries used by UI pages and services.

Data access layer — ALL SQL lives here.

Every public function uses parameterised queries ('?' placeholders). No SQL strings are constructed outside this module.

Changes applied (see fix_specs.json): FIX-001 update_* functions now log FIELD_UPDATE events before returning. FIX-002 delete_* functions now log DELETE events before executing. FIX-007 'print' added to _STEP_TABLES / _ALLOWED_STEPS; get_print_job added. FIX-009 search_all() replaces in-memory Python filtering. FIX-010 Student CRUD + list_jobs_for_student added. FIX-015 report_jobs() added. FIX-016 Delivery columns added to allowed sets for all update functions. FIX-017 preview_backfill_metadata_keys() added.

add_student(last_name, first_name, school='', grade='', preferred_formats='', notes='')

Create a student record and return the new id.

Source code in accessibility_mgr/db/queries.py

def add_student(
    last_name: str, first_name: str,
    school: str = "", grade: str = "",
    preferred_formats: str = "", notes: str = "",
) -> int:
    """Create a student record and return the new id."""
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO student (last_name,first_name,school,grade,preferred_formats,notes) "
            "VALUES (?,?,?,?,?,?)",
            (last_name, first_name, school, grade, preferred_formats, notes),
        )
        return int(cur.lastrowid)

backfill_metadata_keys(approved_keys)

Normalise and backfill typo'd metadata keys into approved keys (writes to DB).

Source code in accessibility_mgr/db/queries.py

def backfill_metadata_keys(approved_keys: list[str]) -> dict[str, Any]:
    """Normalise and backfill typo'd metadata keys into approved keys (writes to DB)."""
    if not approved_keys:
        return {"updated_rows": 0, "deleted_rows": 0, "mappings": {}, "skipped_keys": []}

    def _norm(k: str) -> str:
        k = k.strip().lower().replace(" ", "_").replace("-", "_")
        return re.sub(r"[^a-z0-9_:]", "", k)

    approved_set = set(approved_keys)
    norm_to_key = {_norm(k): k for k in approved_keys}
    norm_candidates = list(norm_to_key.keys())

    with get_conn() as conn:
        distinct = _rows(conn.execute("SELECT DISTINCT meta_key FROM job_metadata ORDER BY meta_key"))

        mappings: dict[str, str] = {}
        skipped: list[str] = []

        for row in distinct:
            source = row["meta_key"]
            if source in approved_set:
                continue
            nsrc = _norm(source)

            if nsrc in norm_to_key:
                mappings[source] = norm_to_key[nsrc]
                continue

            closest = difflib.get_close_matches(nsrc, norm_candidates, n=1, cutoff=0.8)
            if closest:
                mappings[source] = norm_to_key[closest[0]]
            else:
                skipped.append(source)

        updated_rows = 0
        deleted_rows = 0

        for source, target in mappings.items():
            if source == target:
                continue

            rows = _rows(conn.execute(
                "SELECT id, job_type, job_id, meta_value FROM job_metadata WHERE meta_key=?",
                (source,),
            ))

            for r in rows:
                existing = _rows(conn.execute(
                    "SELECT id, meta_value FROM job_metadata "
                    "WHERE job_type=? AND job_id=? AND meta_key=?",
                    (r["job_type"], r["job_id"], target),
                ))

                if not existing:
                    conn.execute(
                        "UPDATE job_metadata SET meta_key=?, updated_at=datetime('now') WHERE id=?",
                        (target, r["id"]),
                    )
                    updated_rows += 1
                    continue

                tgt_id = existing[0]["id"]
                tgt_val = existing[0].get("meta_value") or ""
                src_val = r.get("meta_value") or ""

                merged = tgt_val
                if src_val and src_val not in tgt_val:
                    merged = f"{tgt_val} | {src_val}" if tgt_val else src_val
                    conn.execute(
                        "UPDATE job_metadata SET meta_value=?, updated_at=datetime('now') WHERE id=?",
                        (merged, tgt_id),
                    )
                    updated_rows += 1

                conn.execute("DELETE FROM job_metadata WHERE id=?", (r["id"],))
                deleted_rows += 1

        return {
            "updated_rows": updated_rows,
            "deleted_rows": deleted_rows,
            "mappings": mappings,
            "skipped_keys": skipped,
        }

complete_step(job_type, job_id, step_key, agent='user')

Mark a workflow step as complete and log a STEP_COMPLETE event.

Source code in accessibility_mgr/db/queries.py

def complete_step(job_type: str, job_id: int, step_key: str, agent: str = "user") -> None:
    """Mark a workflow step as complete and log a STEP_COMPLETE event."""
    table = _STEP_TABLES.get(job_type)
    if not table or step_key not in _ALLOWED_STEPS.get(job_type, []):
        raise ValueError(f"Unknown step '{step_key}' for job type '{job_type}'")
    step_date_col = f"{step_key}_date"
    with get_conn() as conn:
        conn.execute(
            f"UPDATE {table} SET {step_key} = 1, {step_date_col} = datetime('now'), updated_at = datetime('now') WHERE id = ?",  # noqa: S608 - table/step/date columns come from fixed maps, never raw user SQL.
            (job_id,),
        )
    log_event(job_type, job_id, "STEP_COMPLETE", "SUCCESS",
              step_key=step_key, agent=agent, detail=f"Step '{step_key}' marked complete")

count_jobs_for_students(student_ids)

Return total job counts for a batch of students in a single query.

Issues one SQL UNION ALL query rather than four separate queries per student. Returns {student_id: total_count}.

Source code in accessibility_mgr/db/queries.py

def count_jobs_for_students(student_ids: list[int]) -> dict[int, int]:
    """Return total job counts for a batch of students in a single query.

    Issues one SQL UNION ALL query rather than four separate queries per
    student. Returns {student_id: total_count}.
    """
    if not student_ids:
        return {}
    placeholders = ",".join("?" * len(student_ids))
    with get_conn() as conn:
        rows = conn.execute(
            f"""
            SELECT student_id, COUNT(*) AS cnt FROM (
                SELECT student_id FROM braille_job         WHERE student_id IN ({placeholders})
                UNION ALL
                SELECT student_id FROM lp_ebraille_job     WHERE student_id IN ({placeholders})
                UNION ALL
                SELECT student_id FROM tactile_graphics_job WHERE student_id IN ({placeholders})
                UNION ALL
                SELECT student_id FROM print_job            WHERE student_id IN ({placeholders})
            ) GROUP BY student_id
            """,  # noqa: S608 - placeholders are '?' only; no user content in SQL.
            student_ids * 4,
        ).fetchall()
    return {row[0]: row[1] for row in rows}

deduct_filament(row_id, grams)

Deduct grams from filament stock.

FUN-011: grams must be strictly positive. Zero is a no-op; negative values would silently add stock (MAX(0, qty - negative) = qty + |negative|).

Source code in accessibility_mgr/db/queries.py

def deduct_filament(row_id: int, grams: float) -> None:
    """Deduct *grams* from filament stock.

    FUN-011: grams must be strictly positive.  Zero is a no-op; negative values
    would silently *add* stock (MAX(0, qty - negative) = qty + |negative|).
    """
    if grams <= 0:
        raise ValueError(f"grams must be positive, got {grams!r}")
    with get_conn() as conn:
        conn.execute(
            "UPDATE filament SET quantity_g = MAX(0, quantity_g - ?), "
            "updated_at = datetime('now') WHERE id = ?",
            (grams, row_id),
        )

delete_braille_job(row_id)

Delete a braille job, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_braille_job(row_id: int) -> None:
    """Delete a braille job, logging a DELETE audit event first (FIX-002)."""
    old = get_braille_job(row_id)
    if old:
        log_event(
            "braille", row_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"Job deleted: {old.get('title', '')}",
            extra_metadata={k: old.get(k) for k in
                            ["title", "braille_type", "requester", "priority", "created_at"]},
        )
    with get_conn() as conn:
        conn.execute("DELETE FROM braille_job WHERE id = ?", (row_id,))
        _delete_job_orphans(conn, "braille", row_id)

delete_file_object(file_id)

Delete a file object, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_file_object(file_id: int) -> None:
    """Delete a file object, logging a DELETE audit event first (FIX-002)."""
    row = get_file_object(file_id)
    if row:
        # FIX-002: log before deleting
        log_event(
            "file", file_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"File object deleted: {row.get('original_name', '')}",
            extra_metadata={k: row.get(k) for k in
                            ["original_name", "stored_path", "checksum_sha256", "file_use"]},
        )
        _sp = Path(row["stored_path"])
        stored = _sp if _sp.is_absolute() else FILES_DIR / _sp
        stored = stored.resolve()
        # SEC-004: only unlink if the resolved path is inside ARTIFACTS_DIR or FILES_DIR
        safe_roots = (ARTIFACTS_DIR.resolve(), FILES_DIR.resolve())
        if any(str(stored).startswith(str(root)) for root in safe_roots):
            stored.unlink(missing_ok=True)
        else:
            import logging as _log
            _log.getLogger(__name__).warning(
                "delete_file_object: refusing to unlink '%s' outside permitted dirs", stored
            )
    with get_conn() as conn:
        conn.execute("DELETE FROM file_object WHERE id = ?", (file_id,))

delete_lp_job(row_id)

Delete an LP/eBraille job, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_lp_job(row_id: int) -> None:
    """Delete an LP/eBraille job, logging a DELETE audit event first (FIX-002)."""
    old = get_lp_job(row_id)
    if old:
        log_event(
            "lp_ebraille", row_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"Job deleted: {old.get('title', '')}",
            extra_metadata={k: old.get(k) for k in
                            ["title", "job_type", "requester", "priority", "created_at"]},
        )
    with get_conn() as conn:
        conn.execute("DELETE FROM lp_ebraille_job WHERE id = ?", (row_id,))
        _delete_job_orphans(conn, "lp_ebraille", row_id)

delete_material_category(row_id)

Soft-delete a material category (sets active=0).

Source code in accessibility_mgr/db/queries.py

def delete_material_category(row_id: int) -> None:
    """Soft-delete a material category (sets active=0)."""
    set_material_category_active(row_id, 0)

delete_print_job(row_id)

Delete a print job, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_print_job(row_id: int) -> None:
    """Delete a print job, logging a DELETE audit event first (FIX-002)."""
    old = get_print_job(row_id)
    if old:
        log_event(
            "print", row_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"Print job deleted: {old.get('object_name') or old.get('file_name') or 'unnamed'}",
            extra_metadata={k: old.get(k) for k in
                            ["object_name", "printer_id", "filament_used_g", "successful", "printed_at"]},
        )
    with get_conn() as conn:
        conn.execute("DELETE FROM print_job WHERE id = ?", (row_id,))
        _delete_job_orphans(conn, "print", row_id)

delete_student(student_id)

Soft-delete a student (sets active=0).

Source code in accessibility_mgr/db/queries.py

def delete_student(student_id: int) -> None:
    """Soft-delete a student (sets active=0)."""
    update_student(student_id, active=0)

delete_tactile_job(row_id)

Delete a tactile job, logging a DELETE audit event first (FIX-002).

Source code in accessibility_mgr/db/queries.py

def delete_tactile_job(row_id: int) -> None:
    """Delete a tactile job, logging a DELETE audit event first (FIX-002)."""
    old = get_tactile_job(row_id)
    if old:
        log_event(
            "tactile", row_id, "DELETE", "SUCCESS",
            agent="system",
            detail=f"Job deleted: {old.get('title', '')}",
            extra_metadata={k: old.get(k) for k in
                            ["title", "tactile_type", "requester", "priority", "created_at"]},
        )
    with get_conn() as conn:
        conn.execute("DELETE FROM tactile_graphics_job WHERE id = ?", (row_id,))
        _delete_job_orphans(conn, "tactile", row_id)

delete_workflow_step(row_id)

Soft-delete a workflow step (sets active=0).

Source code in accessibility_mgr/db/queries.py

def delete_workflow_step(row_id: int) -> None:
    """Soft-delete a workflow step (sets active=0)."""
    set_workflow_step_active(row_id, 0)

get_print_job(row_id)

Fetch a single print job by id (FIX-001: needed for pre-update snapshot).

Source code in accessibility_mgr/db/queries.py

def get_print_job(row_id: int) -> Optional[dict[str, Any]]:
    """Fetch a single print job by id (FIX-001: needed for pre-update snapshot)."""
    with get_conn() as conn:
        rows = _rows(conn.execute("""
            SELECT pj.*,
                   p.name  AS printer_name,
                   f.brand || ' ' || f.color || ' ' || f.filament_type AS filament_desc
            FROM print_job pj
            LEFT JOIN printer  p ON p.id = pj.printer_id
            LEFT JOIN filament f ON f.id = pj.filament_id
            WHERE pj.id = ?
        """, (row_id,)))
        return rows[0] if rows else None

get_student(student_id)

Fetch a single student record by id.

Source code in accessibility_mgr/db/queries.py

def get_student(student_id: int) -> Optional[dict[str, Any]]:
    """Fetch a single student record by id."""
    with get_conn() as conn:
        rows = _rows(conn.execute("SELECT * FROM student WHERE id = ?", (student_id,)))
        return rows[0] if rows else None

ingest_file(source_path, file_use='ORIGINAL', format_name='', format_version='', encoding='', extra_metadata=None, project_title='', student_initials='', school_name='', grade_level='', subject='')

Copy a file into the artifact store, compute SHA-256, and create a file_object record.

SEC-004 / FUN-023: the source_path must exist and must resolve to a path inside FILES_DIR (the staging area). This prevents callers from staging arbitrary files from anywhere on the filesystem into the artifact store.

Source code in accessibility_mgr/db/queries.py

def ingest_file(
    source_path: str,
    file_use: str = "ORIGINAL",
    format_name: str = "",
    format_version: str = "",
    encoding: str = "",
    extra_metadata: Optional[dict[str, Any]] = None,
    project_title: str = "",
    student_initials: str = "",
    school_name: str = "",
    grade_level: str = "",
    subject: str = "",
) -> int:
    """Copy a file into the artifact store, compute SHA-256, and create a file_object record.

    SEC-004 / FUN-023: the *source_path* must exist and must resolve to a path
    inside FILES_DIR (the staging area).  This prevents callers from staging
    arbitrary files from anywhere on the filesystem into the artifact store.
    """
    src = Path(source_path).resolve()
    if not src.exists():
        raise FileNotFoundError(f"Source file not found: {src}")

    # SEC-004: ensure source resolves inside the permitted staging directory
    try:
        src.relative_to(FILES_DIR.resolve())
    except ValueError:
        raise PermissionError(
            f"ingest_file: source '{src}' is outside the permitted staging directory "
            f"'{FILES_DIR}'.  Stage files via the upload handler before ingesting."
        )

    file_uuid = str(uuid.uuid4())

    if project_title:
        safe_project_title = _sanitize_name(project_title) or file_uuid[:8]
        project_dir = ARTIFACTS_DIR / safe_project_title
        project_dir.mkdir(parents=True, exist_ok=True)

        name_parts: list[str] = []
        if student_initials:
            cleaned = _sanitize_name(student_initials)
            if cleaned:
                name_parts.append(cleaned)
        if school_name:
            cleaned = _sanitize_name(school_name)
            if cleaned:
                name_parts.append(cleaned)
        if grade_level:
            cleaned = _sanitize_name(grade_level)
            if cleaned:
                name_parts.append(f"Grade{cleaned}")
        if subject:
            cleaned = _sanitize_name(subject)
            if cleaned:
                name_parts.append(cleaned)

        artifact_stem = "_".join(name_parts) or file_uuid[:8]
        max_path_len = 240

        # Keep destination paths safely below typical filesystem path limits.
        while len(str(project_dir / f"{artifact_stem}{src.suffix}")) > max_path_len and len(artifact_stem) > 16:
            artifact_stem = artifact_stem[:-8]

        if len(f"{artifact_stem}{src.suffix}") > 255:
            raise ValueError("Artifact file name exceeds filesystem component limits")

        dest = project_dir / f"{artifact_stem}{src.suffix}"
        if len(str(dest)) > max_path_len:
            raise ValueError(
                "Artifact destination path is too long; shorten project or metadata values"
            )
        if dest.exists():
            dest = project_dir / f"{artifact_stem}_{file_uuid[:8]}{src.suffix}"
            if len(str(dest)) > max_path_len:
                raise ValueError(
                    "Artifact destination path is too long after collision handling"
                )
        stored_path_val = str(dest)
    else:
        dest = FILES_DIR / f"{file_uuid}{src.suffix}"
        stored_path_val = dest.name

    shutil.copy2(src, dest)
    checksum = _sha256(dest)
    size_bytes = dest.stat().st_size
    mime_type = mimetypes.guess_type(src.name)[0] or "application/octet-stream"

    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO file_object (uuid,original_name,stored_path,mime_type,size_bytes,"
            "checksum_sha256,file_use,format_name,format_version,encoding,extra_metadata) "
            "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
            (file_uuid, src.name, stored_path_val, mime_type, size_bytes, checksum,
             file_use, format_name, format_version, encoding,
             json.dumps(extra_metadata) if extra_metadata else None),
        )
        return int(cur.lastrowid)

list_distinct_metadata_keys()

Return all metadata keys in use with occurrence counts.

Source code in accessibility_mgr/db/queries.py

def list_distinct_metadata_keys() -> list[dict[str, Any]]:
    """Return all metadata keys in use with occurrence counts."""
    with get_conn() as conn:
        return _rows(conn.execute(
            "SELECT meta_key, COUNT(*) AS usage_count "
            "FROM job_metadata GROUP BY meta_key ORDER BY usage_count DESC, meta_key"
        ))

list_filaments()

Return all filament records ordered by brand and colour.

Source code in accessibility_mgr/db/queries.py

def list_filaments() -> list[dict[str, Any]]:
    """Return all filament records ordered by brand and colour."""
    with get_conn() as conn:
        return _rows(conn.execute("SELECT * FROM filament ORDER BY brand, color"))

list_jobs_for_student(student_id)

Return all jobs linked to a student, grouped by type.

Source code in accessibility_mgr/db/queries.py

def list_jobs_for_student(student_id: int) -> dict[str, list[dict[str, Any]]]:
    """Return all jobs linked to a student, grouped by type."""
    with get_conn() as conn:
        braille = _rows(conn.execute(
            "SELECT *, 'braille' AS job_type FROM braille_job WHERE student_id = ? ORDER BY created_at DESC",
            (student_id,),
        ))
        lp = _rows(conn.execute(
            "SELECT *, job_type FROM lp_ebraille_job WHERE student_id = ? ORDER BY created_at DESC",
            (student_id,),
        ))
        tactile = _rows(conn.execute(
            "SELECT *, 'tactile' AS job_type FROM tactile_graphics_job WHERE student_id = ? ORDER BY created_at DESC",
            (student_id,),
        ))
        print_jobs = _rows(conn.execute(
            "SELECT *, 'print' AS job_type FROM print_job WHERE student_id = ? ORDER BY printed_at DESC",
            (student_id,),
        ))
    return {
        "braille": braille,
        "lp_ebraille": lp,
        "tactile": tactile,
        "print": print_jobs,
    }

list_students(active_only=True)

Return student records ordered by last name, first name.

Source code in accessibility_mgr/db/queries.py

def list_students(active_only: bool = True) -> list[dict[str, Any]]:
    """Return student records ordered by last name, first name."""
    with get_conn() as conn:
        where = "WHERE active = 1" if active_only else ""
        return _rows(conn.execute(
            f"SELECT * FROM student {where} ORDER BY last_name, first_name"  # noqa: S608 - where clause is a fixed toggle, no user-supplied SQL.
        ))

list_students_page(search=None, active_only=True, limit=50, offset=0)

Return a paginated page of student records with optional search.

Source code in accessibility_mgr/db/queries.py

def list_students_page(
    search: Optional[str] = None,
    active_only: bool = True,
    limit: int = 50,
    offset: int = 0,
) -> list[dict[str, Any]]:
    """Return a paginated page of student records with optional search."""
    filters: list[str] = []
    params: list[Any] = []
    if active_only:
        filters.append("active = 1")
    if search:
        term = f"%{search}%"
        filters.append("(last_name LIKE ? OR first_name LIKE ? OR school LIKE ?)")
        params.extend([term, term, term])
    where = ("WHERE " + " AND ".join(filters)) if filters else ""
    params.extend([limit, max(0, offset)])
    with get_conn() as conn:
        return _rows(conn.execute(
            f"SELECT * FROM student {where} ORDER BY last_name, first_name LIMIT ? OFFSET ?",  # noqa: S608 - where clause from fixed SQL fragments; values parameterised.
            params,
        ))

log_qa_measure(*, engine, epub_path, passed, score, error_count=0, warning_count=0, info_count=0, issues=None, job_type=None, job_id=None, reviewer=None, reviewer_notes='', checked_at=None)

Persist a reviewer-submitted QA measure and return its row id.

When the measure is linked to a job (job_type + job_id), this also writes a QA_MEASURE_SUBMITTED entry to that job's event log so the result is visible alongside the job's other history, consistent with how log_qa_run results are linked via FIX-012 in ui/qa.py.

Source code in accessibility_mgr/db/queries.py

def log_qa_measure(
    *,
    engine: str,
    epub_path: str,
    passed: bool,
    score: int,
    error_count: int = 0,
    warning_count: int = 0,
    info_count: int = 0,
    issues: Optional[list[dict[str, Any]]] = None,
    job_type: Optional[str] = None,
    job_id: Optional[int] = None,
    reviewer: Optional[str] = None,
    reviewer_notes: str = "",
    checked_at: Optional[str] = None,
) -> int:
    """Persist a reviewer-submitted QA measure and return its row id.

    When the measure is linked to a job (job_type + job_id), this also
    writes a QA_MEASURE_SUBMITTED entry to that job's event log so the
    result is visible alongside the job's other history, consistent with
    how log_qa_run results are linked via FIX-012 in ui/qa.py.
    """
    with get_conn() as conn:
        cur = conn.execute(
            "INSERT INTO qa_measure (engine,epub_path,job_type,job_id,passed,"
            "score,error_count,warning_count,info_count,issues,reviewer,"
            "reviewer_notes,checked_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
            (
                engine, epub_path, job_type, job_id, 1 if passed else 0, score,
                error_count, warning_count, info_count,
                json.dumps(issues or []), reviewer, reviewer_notes, checked_at,
            ),
        )
        measure_id = int(cur.lastrowid)

    if job_type and job_id:
        log_event(
            job_type, job_id, "QA_MEASURE_SUBMITTED",
            event_outcome="SUCCESS" if passed else "FAILURE",
            agent=reviewer or "system",
            detail=(
                f"{engine} accessibility QA — score {score}, "
                f"{error_count} error(s), {warning_count} warning(s)"
            ),
            extra_metadata={"qa_measure_id": measure_id, "score": score},
        )

    return measure_id

preview_backfill_metadata_keys(approved_keys)

Return proposed key mappings without writing to the database (FIX-017 dry-run).

Source code in accessibility_mgr/db/queries.py

def preview_backfill_metadata_keys(approved_keys: list[str]) -> dict[str, Any]:
    """Return proposed key mappings without writing to the database (FIX-017 dry-run)."""
    if not approved_keys:
        return {"mappings": {}, "skipped_keys": [], "usage_counts": {}}

    def _norm(k: str) -> str:
        k = k.strip().lower().replace(" ", "_").replace("-", "_")
        return re.sub(r"[^a-z0-9_:]", "", k)

    approved_set = set(approved_keys)
    norm_to_key = {_norm(k): k for k in approved_keys}
    norm_candidates = list(norm_to_key.keys())

    with get_conn() as conn:
        distinct = _rows(conn.execute(
            "SELECT meta_key, COUNT(*) AS usage_count FROM job_metadata "
            "GROUP BY meta_key ORDER BY meta_key"
        ))

    mappings: dict[str, str] = {}
    skipped: list[str] = []
    usage_counts: dict[str, int] = {r["meta_key"]: r["usage_count"] for r in distinct}

    for row in distinct:
        source = row["meta_key"]
        if source in approved_set:
            continue
        nsrc = _norm(source)

        if nsrc in norm_to_key:
            mappings[source] = norm_to_key[nsrc]
            continue

        closest = difflib.get_close_matches(nsrc, norm_candidates, n=1, cutoff=0.8)
        if closest:
            mappings[source] = norm_to_key[closest[0]]
        else:
            skipped.append(source)

    return {
        "mappings": mappings,
        "skipped_keys": skipped,
        "usage_counts": usage_counts,
    }

record_delivery(job_type, job_id, delivery_method, delivery_recipient, delivery_date=None, delivery_notes='', agent='user')

Record delivery details, complete the 'delivered' step, and log a DELIVERY event (FIX-016).

Source code in accessibility_mgr/db/queries.py

def record_delivery(
    job_type: str, job_id: int,
    delivery_method: str,
    delivery_recipient: str,
    delivery_date: Optional[str] = None,
    delivery_notes: str = "",
    agent: str = "user",
) -> None:
    """Record delivery details, complete the 'delivered' step, and log a DELIVERY event (FIX-016)."""
    from datetime import date
    d_date = delivery_date or date.today().isoformat()

    # Update delivery columns directly without emitting a redundant FIELD_UPDATE event.
    _table_map = {
        "braille":     "braille_job",
        "lp_ebraille": "lp_ebraille_job",
        "tactile":     "tactile_graphics_job",
        "print":       "print_job",
    }
    table = _table_map.get(job_type)
    if table and table in _SAFE_TABLES:
        with get_conn() as conn:
            conn.execute(  # noqa: S608 - table comes from a fixed allow-map, not user input.
                f"UPDATE {table} SET delivered=1, delivery_date=?, delivery_method=?, "
                f"delivery_recipient=?, delivery_notes=?, updated_at=datetime('now') WHERE id=?",
                (d_date, delivery_method, delivery_recipient, delivery_notes, job_id),
            )

    log_event(
        job_type, job_id, "DELIVERY", "SUCCESS",
        step_key="delivered",
        agent=agent,
        detail=f"Delivered to {delivery_recipient} via {delivery_method} on {d_date}",
        extra_metadata={
            "delivery_method": delivery_method,
            "delivery_recipient": delivery_recipient,
            "delivery_date": d_date,
            "delivery_notes": delivery_notes,
        },
    )

report_jobs(school=None, grade=None, job_type=None, status=None, priority=None, date_from=None, date_to=None, student_id=None)

Return filtered job lists grouped by type with summary counts.

Parameters

school : str, optional Filter by student.school (exact) or dc:coverage metadata value (LIKE). grade : str, optional Filter by student.grade or grade_level metadata value (LIKE). job_type : str, optional One of 'braille', 'lp_ebraille', 'tactile', 'print'. If None, all types returned. status : str, optional 'not_started', 'in_progress', or 'delivered'. priority : str, optional One of 'low', 'normal', 'high', 'urgent'. date_from : str, optional ISO date string — jobs created on or after this date. date_to : str, optional ISO date string — jobs created on or before this date. student_id : int, optional Return only jobs for this student.

Source code in accessibility_mgr/db/queries.py

def report_jobs(
    school: Optional[str] = None,
    grade: Optional[str] = None,
    job_type: Optional[str] = None,
    status: Optional[str] = None,
    priority: Optional[str] = None,
    date_from: Optional[str] = None,
    date_to: Optional[str] = None,
    student_id: Optional[int] = None,
) -> dict[str, Any]:
    """Return filtered job lists grouped by type with summary counts.

    Parameters
    ----------
    school : str, optional
        Filter by student.school (exact) or dc:coverage metadata value (LIKE).
    grade : str, optional
        Filter by student.grade or grade_level metadata value (LIKE).
    job_type : str, optional
        One of 'braille', 'lp_ebraille', 'tactile', 'print'. If None, all types returned.
    status : str, optional
        'not_started', 'in_progress', or 'delivered'.
    priority : str, optional
        One of 'low', 'normal', 'high', 'urgent'.
    date_from : str, optional
        ISO date string — jobs created on or after this date.
    date_to : str, optional
        ISO date string — jobs created on or before this date.
    student_id : int, optional
        Return only jobs for this student.
    """

    def _step_status_expr(steps: list[str]) -> str:
        """Build SQL CASE expression deriving a status label from step columns."""
        delivered_col = steps[-1]
        any_done = " + ".join(f"COALESCE({s}, 0)" for s in steps)
        return (
            f"CASE WHEN {delivered_col} = 1 THEN 'delivered' "
            f"WHEN ({any_done}) > 0 THEN 'in_progress' "
            f"ELSE 'not_started' END"
        )

    def _run(
        table: str,
        type_label: str,
        steps: list[str],
        title_col: str = "title",
        date_col: str = "created_at",
        priority_expr: str = "j.priority",
        type_expr: str | None = None,
    ) -> list[dict[str, Any]]:
        # SAFE: all SQL-interpolated identifiers come from fixed in-function constants,
        # not user input. User-supplied values are bound via '?' parameterisation only.
        filters: list[str] = []
        params: list[Any] = []

        if student_id is not None:
            filters.append("j.student_id = ?")
            params.append(student_id)
        if date_from:
            filters.append(f"j.{date_col} >= ?")
            params.append(date_from)
        if date_to:
            filters.append(f"j.{date_col} <= ?")
            params.append(date_to)
        if school:
            filters.append(
                "(s.school LIKE ? OR EXISTS ("
                "  SELECT 1 FROM job_metadata jm "
                "  WHERE jm.job_type=? AND jm.job_id=j.id "
                "  AND jm.meta_key='dc:coverage' AND jm.meta_value LIKE ?"
                "))"
            )
            params += [f"%{school}%", type_label, f"%{school}%"]
        if grade:
            filters.append(
                "(s.grade LIKE ? OR EXISTS ("
                "  SELECT 1 FROM job_metadata jm "
                "  WHERE jm.job_type=? AND jm.job_id=j.id "
                "  AND jm.meta_key='grade_level' AND jm.meta_value LIKE ?"
                "))"
            )
            params += [f"%{grade}%", type_label, f"%{grade}%"]

        status_expr = _step_status_expr(steps)
        if status:
            filters.append(f"({status_expr}) = ?")
            params.append(status)
        if priority and priority_expr != "NULL":
            filters.append(f"{priority_expr} = ?")
            params.append(priority)

        where = ("WHERE " + " AND ".join(filters)) if filters else ""
        resolved_type_expr = type_expr or f"'{type_label}'"

        sql = f"""
                                 SELECT j.id, j.{title_col} AS title, {resolved_type_expr} AS job_type,
                 j.requester, {priority_expr} AS priority, j.{date_col} AS created_at,
                   s.last_name, s.first_name, s.school, s.grade,
                   ({status_expr}) AS status
            FROM {table} j
            LEFT JOIN student s ON s.id = j.student_id
            {where}
                 ORDER BY j.{date_col} DESC
        """  # noqa: S608 - table/column identifiers come from trusted in-module constants.
        with get_conn() as conn:
            return _rows(conn.execute(sql, params))

    b_steps   = ["digitized", "formatted", "brailled", "proofread", "delivered"]
    lp_steps  = ["digitized", "formatted", "converted", "proofread", "delivered"]
    tac_steps = ["designed", "produced", "qa_reviewed", "delivered"]
    prt_steps = ["designed", "sliced", "printed", "inspected", "delivered"]

    results: dict[str, list[dict[str, Any]]] = {}

    if job_type in (None, "braille"):
        results["braille"] = _run("braille_job", "braille", b_steps)
    if job_type in (None, "lp_ebraille"):
        results["lp_ebraille"] = _run(
            "lp_ebraille_job",
            "lp_ebraille",
            lp_steps,
            type_expr="COALESCE(j.job_type, 'lp_ebraille')",
        )
    if job_type in (None, "tactile"):
        results["tactile"] = _run("tactile_graphics_job", "tactile", tac_steps)
    if job_type in (None, "print"):
        results["print"] = _run(
            "print_job",
            "print",
            prt_steps,
            title_col="object_name",
            date_col="printed_at",
            priority_expr="'normal'",
        )

    all_jobs: list[dict[str, Any]] = []
    for rows in results.values():
        all_jobs.extend(rows)

    by_type = {k: len(v) for k, v in results.items()}
    return {
        "total_jobs": len(all_jobs),
        "by_type": by_type,
        "jobs": all_jobs,
        "by_type_lists": results,
    }

revert_step(job_type, job_id, step_key, agent='user', reason='')

Revert a workflow step and log a STEP_REVERT event.

Source code in accessibility_mgr/db/queries.py

def revert_step(
    job_type: str, job_id: int, step_key: str,
    agent: str = "user", reason: str = "",
) -> None:
    """Revert a workflow step and log a STEP_REVERT event."""
    table = _STEP_TABLES.get(job_type)
    if not table or step_key not in _ALLOWED_STEPS.get(job_type, []):
        raise ValueError(f"Unknown step '{step_key}' for job type '{job_type}'")
    step_date_col = f"{step_key}_date"
    with get_conn() as conn:
        conn.execute(
            f"UPDATE {table} SET {step_key} = 0, {step_date_col} = NULL, updated_at = datetime('now') WHERE id = ?",  # noqa: S608 - table/step/date columns come from fixed maps, never raw user SQL.
            (job_id,),
        )
    log_event(job_type, job_id, "STEP_REVERT", "WARNING",
              step_key=step_key, agent=agent,
              detail=f"Step '{step_key}' reverted" + (f": {reason}" if reason else ""))

search_all(query, limit=200)

Search all job tables, files, metadata, and event log using SQL LIKE queries.

Replaces the in-memory Python filtering in the UI layer (FIX-009). Also searches event log detail text (FIX-014) and file checksums.

Parameters

query : str The search term. Applied with LIKE '%term%' across text columns. If exactly 64 hex characters, also tested as an exact SHA-256 match. limit : int Maximum rows returned per result category.

Source code in accessibility_mgr/db/queries.py

def search_all(query: str, limit: int = 200) -> dict[str, list[dict[str, Any]]]:
    """Search all job tables, files, metadata, and event log using SQL LIKE queries.

    Replaces the in-memory Python filtering in the UI layer (FIX-009).
    Also searches event log detail text (FIX-014) and file checksums.

    Parameters
    ----------
    query : str
        The search term. Applied with LIKE '%term%' across text columns.
        If exactly 64 hex characters, also tested as an exact SHA-256 match.
    limit : int
        Maximum rows returned per result category.
    """
    term = f"%{query}%"
    # SHA-256 exact match for 64-char hex strings
    is_checksum = len(query) == 64 and all(c in "0123456789abcdefABCDEF" for c in query)

    with get_conn() as conn:
        if not is_checksum:
            try:
                def _fts_ids(table: str) -> list[int]:
                    rows = conn.execute(
                        f"SELECT id FROM {table} WHERE {table} MATCH ? LIMIT ?",  # noqa: S608 - table comes from fixed in-function constants.
                        (query, limit),
                    ).fetchall()
                    return [int(r[0]) for r in rows]

                def _rows_by_ids(
                    table: str,
                    columns: str,
                    order_by: str,
                    ids: list[int],
                ) -> list[dict[str, Any]]:
                    if not ids:
                        return []
                    placeholders = ",".join("?" for _ in ids)
                    sql = (
                        f"SELECT {columns} FROM {table} "
                        f"WHERE id IN ({placeholders}) ORDER BY {order_by} LIMIT ?"
                    )
                    return _rows(conn.execute(sql, [*ids, limit]))

                braille_jobs = _rows_by_ids(
                    "braille_job",
                    "id, title, braille_type, requester, priority, created_at",
                    "created_at DESC",
                    _fts_ids("braille_job_fts"),
                )
                lp_jobs = _rows_by_ids(
                    "lp_ebraille_job",
                    "id, title, job_type, requester, priority, created_at",
                    "created_at DESC",
                    _fts_ids("lp_ebraille_job_fts"),
                )
                tactile_jobs = _rows_by_ids(
                    "tactile_graphics_job",
                    "id, title, tactile_type, requester, priority, created_at",
                    "created_at DESC",
                    _fts_ids("tactile_graphics_job_fts"),
                )
                print_jobs = _rows_by_ids(
                    "print_job",
                    "id, object_name, file_name, requester, successful, printed_at",
                    "printed_at DESC",
                    _fts_ids("print_job_fts"),
                )
                files = _rows_by_ids(
                    "file_object",
                    "id, original_name, stored_path, file_use, format_name, checksum_sha256, created_at",
                    "created_at DESC",
                    _fts_ids("file_object_fts"),
                )
                metadata = _rows(conn.execute(
                    "SELECT job_type, CAST(job_id AS INTEGER) AS job_id, meta_key, meta_value "
                    "FROM job_metadata_fts WHERE job_metadata_fts MATCH ? LIMIT ?",
                    (query, limit),
                ))
                events = _rows(conn.execute(
                    "SELECT id, job_type, CAST(job_id AS INTEGER) AS job_id, event_type, agent, detail "
                    "FROM metadata_event_fts WHERE metadata_event_fts MATCH ? LIMIT ?",
                    (query, limit),
                ))
                students = _rows(conn.execute(
                    "SELECT id, last_name, first_name, school, grade "
                    "FROM student WHERE last_name LIKE ? OR first_name LIKE ? "
                    "OR school LIKE ? OR notes LIKE ? ORDER BY last_name LIMIT ?",
                    (term, term, term, term, limit),
                ))

                return {
                    "braille_jobs": braille_jobs,
                    "lp_jobs": lp_jobs,
                    "tactile_jobs": tactile_jobs,
                    "print_jobs": print_jobs,
                    "files": files,
                    "metadata": metadata,
                    "events": events,
                    "students": students,
                }
            except sqlite3.OperationalError:
                # Fallback for environments where FTS tables are unavailable.
                pass

        braille_jobs = _rows(conn.execute(
            "SELECT id, title, braille_type, requester, priority, created_at "
            "FROM braille_job WHERE title LIKE ? OR requester LIKE ? OR notes LIKE ? "
            "ORDER BY created_at DESC LIMIT ?",
            (term, term, term, limit),
        ))

        lp_jobs = _rows(conn.execute(
            "SELECT id, title, job_type, requester, priority, created_at "
            "FROM lp_ebraille_job WHERE title LIKE ? OR requester LIKE ? OR notes LIKE ? "
            "ORDER BY created_at DESC LIMIT ?",
            (term, term, term, limit),
        ))

        tactile_jobs = _rows(conn.execute(
            "SELECT id, title, tactile_type, requester, priority, created_at "
            "FROM tactile_graphics_job WHERE title LIKE ? OR requester LIKE ? OR notes LIKE ? "
            "ORDER BY created_at DESC LIMIT ?",
            (term, term, term, limit),
        ))

        print_jobs = _rows(conn.execute(
            "SELECT id, object_name, file_name, requester, successful, printed_at "
            "FROM print_job WHERE object_name LIKE ? OR requester LIKE ? "
            "OR file_name LIKE ? OR notes LIKE ? ORDER BY printed_at DESC LIMIT ?",
            (term, term, term, term, limit),
        ))

        if is_checksum:
            files = _rows(conn.execute(
                "SELECT id, original_name, stored_path, file_use, format_name, "
                "checksum_sha256, created_at FROM file_object "
                "WHERE original_name LIKE ? OR stored_path LIKE ? OR format_name LIKE ? "
                "OR encoding LIKE ? OR checksum_sha256 = ? ORDER BY created_at DESC LIMIT ?",
                (term, term, term, term, query, limit),
            ))
        else:
            files = _rows(conn.execute(
                "SELECT id, original_name, stored_path, file_use, format_name, "
                "checksum_sha256, created_at FROM file_object "
                "WHERE original_name LIKE ? OR stored_path LIKE ? OR format_name LIKE ? "
                "OR encoding LIKE ? ORDER BY created_at DESC LIMIT ?",
                (term, term, term, term, limit),
            ))

        metadata = _rows(conn.execute(
            "SELECT jm.job_type, jm.job_id, jm.meta_key, jm.meta_value "
            "FROM job_metadata jm WHERE jm.meta_key LIKE ? OR jm.meta_value LIKE ? "
            "ORDER BY jm.job_type, jm.job_id LIMIT ?",
            (term, term, limit),
        ))

        # FIX-014: event log search
        events = _rows(conn.execute(
            "SELECT me.id, me.job_type, me.job_id, me.event_type, me.agent, "
            "me.detail, me.event_datetime "
            "FROM metadata_event me "
            "WHERE me.detail LIKE ? OR me.agent LIKE ? OR me.event_type LIKE ? "
            "ORDER BY me.event_datetime DESC LIMIT ?",
            (term, term, term, limit),
        ))

        students = _rows(conn.execute(
            "SELECT id, last_name, first_name, school, grade "
            "FROM student WHERE last_name LIKE ? OR first_name LIKE ? "
            "OR school LIKE ? OR notes LIKE ? ORDER BY last_name LIMIT ?",
            (term, term, term, term, limit),
        ))

    return {
        "braille_jobs": braille_jobs,
        "lp_jobs": lp_jobs,
        "tactile_jobs": tactile_jobs,
        "print_jobs": print_jobs,
        "files": files,
        "metadata": metadata,
        "events": events,
        "students": students,
    }

update_braille_job(row_id, **fields)

Update a braille job and log a FIELD_UPDATE audit event (FIX-001, FIX-016).

Source code in accessibility_mgr/db/queries.py

def update_braille_job(row_id: int, **fields: Any) -> None:
    """Update a braille job and log a FIELD_UPDATE audit event (FIX-001, FIX-016)."""
    old = get_braille_job(row_id)
    allowed = {
        "title", "braille_type", "embosser_id", "requester", "request_date", "due_date",
        "priority", "digitized", "formatted", "brailled", "proofread", "delivered",
        "notes", "student_id",
        # FIX-016: delivery columns
        "delivery_date", "delivery_method", "delivery_recipient", "delivery_notes",
    }
    sql, vals = _build_update_sql("braille_job", fields, allowed)
    with get_conn() as conn:
        conn.execute(sql, vals + [row_id])
    log_event(
        "braille", row_id, "FIELD_UPDATE", "SUCCESS",
        agent="system",
        detail=f"Updated fields: {list(fields.keys())}",
        extra_metadata={
            "changed_fields": list(fields.keys()),
            "previous_values": {k: old.get(k) for k in fields if old},
        },
    )

update_lp_job(row_id, **fields)

Update an LP/eBraille job and log a FIELD_UPDATE audit event (FIX-001, FIX-016).

Source code in accessibility_mgr/db/queries.py

def update_lp_job(row_id: int, **fields: Any) -> None:
    """Update an LP/eBraille job and log a FIELD_UPDATE audit event (FIX-001, FIX-016)."""
    old = get_lp_job(row_id)
    allowed = {
        "title", "job_type", "requester", "request_date", "due_date",
        "priority", "digitized", "formatted", "converted", "proofread",
        "delivered", "notes", "student_id",
        # FIX-016
        "delivery_date", "delivery_method", "delivery_recipient", "delivery_notes",
    }
    sql, vals = _build_update_sql("lp_ebraille_job", fields, allowed)
    with get_conn() as conn:
        conn.execute(sql, vals + [row_id])
    log_event(
        "lp_ebraille", row_id, "FIELD_UPDATE", "SUCCESS",
        agent="system",
        detail=f"Updated fields: {list(fields.keys())}",
        extra_metadata={
            "changed_fields": list(fields.keys()),
            "previous_values": {k: old.get(k) for k in fields if old},
        },
    )

update_print_job(row_id, **fields)

Update a print job and log a FIELD_UPDATE audit event (FIX-001, FIX-016).

Source code in accessibility_mgr/db/queries.py

def update_print_job(row_id: int, **fields: Any) -> None:
    """Update a print job and log a FIELD_UPDATE audit event (FIX-001, FIX-016)."""
    # FIX-001: snapshot before update
    old = get_print_job(row_id)
    allowed = {
        "printer_id", "filament_id", "filament_used_g", "successful",
        "failure_reason", "object_name", "requester", "request_date", "notes",
        "student_id",
        # FIX-007: step columns
        "designed", "sliced", "printed", "inspected", "delivered",
        # FIX-016: delivery columns
        "delivery_date", "delivery_method", "delivery_recipient", "delivery_notes",
    }
    sql, vals = _build_update_sql("print_job", fields, allowed, has_updated_at=True)
    with get_conn() as conn:
        conn.execute(sql, vals + [row_id])
    # FIX-001: log what changed
    log_event(
        "print", row_id, "FIELD_UPDATE", "SUCCESS",
        agent="system",
        detail=f"Updated fields: {list(fields.keys())}",
        extra_metadata={
            "changed_fields": list(fields.keys()),
            "previous_values": {k: old.get(k) for k in fields if old},
        },
    )

update_student(student_id, **fields)

Update a student record.

Source code in accessibility_mgr/db/queries.py

def update_student(student_id: int, **fields: Any) -> None:
    """Update a student record."""
    allowed = {"last_name", "first_name", "school", "grade", "preferred_formats",
               "notes", "active"}
    sql, vals = _build_update_sql("student", fields, allowed)
    with get_conn() as conn:
        conn.execute(sql, vals + [student_id])

update_tactile_job(row_id, **fields)

Update a tactile job and log a FIELD_UPDATE audit event (FIX-001, FIX-016).

Source code in accessibility_mgr/db/queries.py

def update_tactile_job(row_id: int, **fields: Any) -> None:
    """Update a tactile job and log a FIELD_UPDATE audit event (FIX-001, FIX-016)."""
    old = get_tactile_job(row_id)
    allowed = {
        "title", "tactile_type", "requester", "request_date", "due_date",
        "priority", "designed", "produced", "qa_reviewed", "delivered",
        "notes", "student_id",
        # FIX-016
        "delivery_date", "delivery_method", "delivery_recipient", "delivery_notes",
    }
    sql, vals = _build_update_sql("tactile_graphics_job", fields, allowed)
    with get_conn() as conn:
        conn.execute(sql, vals + [row_id])
    log_event(
        "tactile", row_id, "FIELD_UPDATE", "SUCCESS",
        agent="system",
        detail=f"Updated fields: {list(fields.keys())}",
        extra_metadata={
            "changed_fields": list(fields.keys()),
            "previous_values": {k: old.get(k) for k in fields if old},
        },
    )

Seed import

Purpose: CSV normalization and controlled import into inventory tables.

Import inventory seed data from CSV into the project database.

ImportStats dataclass

Counts of inserted and skipped records by inventory category.

Source code in accessibility_mgr/db/seed_import.py

@dataclass
class ImportStats:
    """Counts of inserted and skipped records by inventory category."""
    electronics_added: int = 0
    electronics_skipped: int = 0
    filament_added: int = 0
    filament_skipped: int = 0
    paper_added: int = 0
    paper_skipped: int = 0

import_seed_csv(csv_path, *, replace_existing=False, confirm_replace=True, filament_spool_grams=1000.0, filament_spool_cost=None, dry_run=False)

Import inventory records from a CSV file and return import stats.

Source code in accessibility_mgr/db/seed_import.py

def import_seed_csv(
    csv_path: Path,
    *,
    replace_existing: bool = False,
    confirm_replace: bool = True,
    filament_spool_grams: float = 1000.0,
    filament_spool_cost: float | None = None,
    dry_run: bool = False,
) -> ImportStats:
    """Import inventory records from a CSV file and return import stats."""
    if not csv_path.exists():
        raise FileNotFoundError(f"CSV not found: {csv_path}")

    init_db()
    if replace_existing and not dry_run:
        _replace_existing_inventory(require_confirmation=confirm_replace)

    stats = ImportStats()
    seen_keys: set[str] = set()

    with csv_path.open("r", encoding="utf-8-sig", newline="") as fh:
        reader = csv.DictReader(fh)
        for row in reader:
            category = _clean(row.get("Category"))
            product = _clean(row.get("Product"))
            cost = _clean(row.get("Cost"))
            url = _normalize_url(_clean(row.get("URL")))
            notes = _clean(row.get("Notes"))
            item_type = _clean(row.get("TYPE"))
            brand = _clean(row.get("BRAND"))
            color = _clean(row.get("COLOR"))
            qty_raw = _clean(row.get("QUANTITY"))

            row_key = "|".join(
                [category.lower(), product.lower(), item_type.lower(), brand.lower(), color.lower(), qty_raw, url]
            )
            if row_key in seen_keys:
                continue
            seen_keys.add(row_key)

            if category.lower() == "inventory":
                if item_type.upper() == "BRAILLE_PAPER":
                    paper_type = _paper_type_from_row(product, item_type)
                    qty = _paper_quantity_from_row(product, qty_raw)
                    supplier = brand
                    paper_notes = notes
                    if product:
                        paper_notes = (paper_notes + " | " if paper_notes else "") + f"Product: {product}"

                    if _paper_exists(paper_type, supplier, paper_notes):
                        stats.paper_skipped += 1
                        continue

                    if dry_run:
                        stats.paper_added += 1
                        continue

                    Q.add_paper(
                        paper_type=paper_type,
                        quantity=qty,
                        supplier=supplier,
                        notes=paper_notes,
                    )
                    stats.paper_added += 1
                    continue

                filament_type = _normalize_filament_type(item_type)
                filament_brand = brand or "Unknown"
                filament_color = color or "Unknown"
                spool_count = _to_float(qty_raw)
                quantity_g = (spool_count * filament_spool_grams) if spool_count is not None else 0.0
                cost_per_kg = None
                if filament_spool_cost is not None:
                    cost_per_kg = _cost_per_kg_from_spool(filament_spool_cost, filament_spool_grams)

                if _filament_exists(filament_brand, filament_color, filament_type):
                    stats.filament_skipped += 1
                    continue

                filament_notes = notes
                if spool_count is not None:
                    filament_notes = (
                        (filament_notes + " | ") if filament_notes else ""
                    ) + f"Imported as {spool_count:g} spool(s) @ {filament_spool_grams:g}g each"
                if filament_spool_cost is not None:
                    filament_notes = (
                        (filament_notes + " | ") if filament_notes else ""
                    ) + f"Spool cost used: ${filament_spool_cost:.2f}"

                if dry_run:
                    stats.filament_added += 1
                    continue

                Q.add_filament(
                    brand=filament_brand,
                    color=filament_color,
                    filament_type=filament_type,
                    quantity_g=quantity_g,
                    cost_per_kg=cost_per_kg,
                    supplier="",
                    notes=filament_notes,
                )
                stats.filament_added += 1
                continue

            if not product:
                stats.electronics_skipped += 1
                continue

            elec_category = _normalize_electronics_category(category)
            qty = _to_float(qty_raw) or 0.0
            cost_each = _cost_each(cost)
            supplier = _supplier_from_url(url)
            spec = item_type if item_type else None

            extra_notes = []
            if notes:
                extra_notes.append(notes)
            if url:
                extra_notes.append(f"Source URL: {url}")
            if category:
                extra_notes.append(f"Source category: {category}")
            note_text = " | ".join(extra_notes)

            if _electronics_exists(product, brand, spec or "", supplier):
                stats.electronics_skipped += 1
                continue

            if dry_run:
                stats.electronics_added += 1
                continue

            Q.add_electronic(
                category=elec_category,
                name=product,
                quantity=qty,
                brand=brand or None,
                spec=spec,
                unit="pcs",
                cost_each=cost_each,
                supplier=supplier,
                notes=note_text,
            )
            stats.electronics_added += 1

    return stats

inventory_totals()

Return aggregate row and quantity totals for inventory tables.

Source code in accessibility_mgr/db/seed_import.py

def inventory_totals() -> dict[str, float]:
    """Return aggregate row and quantity totals for inventory tables."""
    with get_conn() as conn:
        electronics_rows = conn.execute("SELECT COUNT(*) AS c, COALESCE(SUM(quantity), 0) AS q FROM electronics").fetchone()
        filament_rows = conn.execute(
            "SELECT COUNT(*) AS c, COALESCE(SUM(quantity_g), 0) AS q, COALESCE(AVG(cost_per_kg), 0) AS avg_cost FROM filament"
        ).fetchone()
        paper_rows = conn.execute("SELECT COUNT(*) AS c, COALESCE(SUM(quantity), 0) AS q FROM braille_paper").fetchone()

    return {
        "electronics_rows": int(electronics_rows["c"]),
        "electronics_quantity_total": float(electronics_rows["q"]),
        "filament_rows": int(filament_rows["c"]),
        "filament_grams_total": float(filament_rows["q"]),
        "filament_avg_cost_per_kg": float(filament_rows["avg_cost"]),
        "paper_rows": int(paper_rows["c"]),
        "paper_quantity_total": float(paper_rows["q"]),
    }

main()

CLI entry point for importing seed inventory data.

Source code in accessibility_mgr/db/seed_import.py

def main() -> None:
    """CLI entry point for importing seed inventory data."""
    args = _build_parser().parse_args()

    if args.verify_only:
        init_db()
        _print_totals()
        return

    spool_grams = args.filament_spool_grams
    if args.grams_per_spool is not None:
        spool_grams = args.grams_per_spool

    stats = import_seed_csv(
        args.csv_path,
        replace_existing=args.replace_existing,
        filament_spool_grams=spool_grams,
        filament_spool_cost=args.filament_spool_cost,
        dry_run=args.dry_run,
    )

    mode = "DRY RUN" if args.dry_run else "IMPORT COMPLETE"
    print(f"[{mode}] {args.csv_path}")
    print(f"Electronics added: {stats.electronics_added}")
    print(f"Electronics skipped: {stats.electronics_skipped}")
    print(f"Filament added: {stats.filament_added}")
    print(f"Filament skipped: {stats.filament_skipped}")
    print(f"Paper added: {stats.paper_added}")
    print(f"Paper skipped: {stats.paper_skipped}")

    if args.verify_totals and not args.dry_run:
        _print_totals()