cass.db¶

SQLite storage: schema and domain models, CRUD, enriched queries, pending-change sync, table catalog metadata.

cass.db ¶

Database layer — schema, CRUD, introspection, and enriched queries.

Import everything from here::

from cass.db import get_db, CanvasStudent, save_canvas_students
from cass import db  # also works

TableCapability `dataclass` ¶

TableCapability(
    editable: bool = False,
    pushable: bool = False,
    pull_guarded: bool = False,
    viewer_rank: int = 999,
    editable_columns: frozenset[str] | None = None,
)

Shared table behavior metadata for CLI and viewer workflows.

QueryResult ¶

QueryResult(
    columns: list[str], rows: list[tuple[object, ...]]
)

Thin wrapper around a sqlite3 cursor for Rich table rendering.

CanvasAssignment ¶

Bases: Struct

A Canvas LMS assignment.

from_api `classmethod` ¶

from_api(
    a: CanvasAssignmentResponse, *, group_name: str = ""
) -> CanvasAssignment

Convert a Canvas API assignment response to a domain model.

CanvasGrade ¶

Bases: Struct

A grade ready for Canvas API push.

CanvasStudent ¶

Bases: Struct

A Canvas LMS student record.

from_api `classmethod` ¶

from_api(
    s: CanvasStudentResponse, *, sis_section_id: str = ""
) -> CanvasStudent

Convert a Canvas API student response to a domain model.

CanvasSubmission ¶

Bases: Struct

A Canvas LMS submission record.

get_table_capability ¶

get_table_capability(table: str) -> TableCapability

Return shared behavior metadata for a table.

connect_db ¶

connect_db(
    root: Path | None = None,
) -> sqlite_utils.Database

Open a lightweight connection to an existing database.

Unlike open_db, this skips schema init and reconciliation — use it when you know the DB already exists (e.g. in a background thread while the viewer's main connection is alive).

db_path ¶

db_path(root: Path | None = None) -> str

Return the SQLite file path.

editable_canvas_tables ¶

editable_canvas_tables() -> set[str]

Return the current editable Canvas-managed working tables.

get_assignment_groups ¶

get_assignment_groups(sdb: Database) -> list[str]

Return distinct non-empty assignment groups, sorted alphabetically.

get_db ¶

get_db(root: Path | None = None) -> sqlite_utils.Database

Return the shared Database, creating it on first call.

get_meta ¶

get_meta(
    key: str, sdb: Database | None = None
) -> str | None

Retrieve a value from the meta table, or None if not found.

load_canvas_assignment_ids ¶

load_canvas_assignment_ids() -> list[int]

Return every Canvas assignment id, ascending.

load_canvas_grades ¶

load_canvas_grades(
    canvas_assignment_id: int | None = None,
) -> list[CanvasGrade]

Load Canvas grades.

load_canvas_student_ids ¶

load_canvas_student_ids() -> set[int]

Return every canvas_id in the Canvas roster.

open_db ¶

open_db(root: Path | None = None) -> sqlite_utils.Database

Open a database for the resolved project root and ensure schema exists.

save_canvas_assignments ¶

save_canvas_assignments(
    assignments: list[CanvasAssignment],
) -> int

Save Canvas assignments from domain models.

save_canvas_grades ¶

save_canvas_grades(grades: list[CanvasGrade]) -> int

Upsert Canvas grades.

save_canvas_students ¶

save_canvas_students(students: list[CanvasStudent]) -> int

Upsert Canvas students into the source table.

save_canvas_submissions ¶

save_canvas_submissions(
    subs: list[CanvasSubmission],
) -> int

Upsert Canvas submissions.

save_meta ¶

save_meta(key: str, value: str) -> None

Store a key-value pair in the meta table.

upsert_canvas_grade ¶

upsert_canvas_grade(
    sdb: Database,
    canvas_user_id: int,
    canvas_assignment_id: int,
    posted_grade: str,
) -> str

Insert or update a single canvas grade, returning the previous value.

Returns:

Type	Description
`str`	Previous posted_grade (empty string if row didn't exist).

get_column_names ¶

get_column_names(sdb: Database, table: str) -> list[str]

Return column names for a table.

get_primary_keys ¶

get_primary_keys(sdb: Database, table: str) -> list[str]

Return primary key column names for a table.

get_tables ¶

get_tables(sdb: Database) -> list[dict[str, str]]

Return list of non-excluded tables with their type.

is_editable ¶

is_editable(sdb: Database, table: str) -> bool

Check if a table is editable (has PKs, is a real table, not read-only).

revert_changes ¶

revert_changes(
    sdb: Database,
    pending: dict[
        str, dict[str, dict[str, dict[str, object]]]
    ],
) -> int

Revert all pending changes in the DB by restoring baseline values.

Returns:

Type	Description
`int`	Number of field changes reverted.

update_cell ¶

update_cell(
    sdb: Database,
    table: str,
    pk: dict[str, object],
    column: str,
    value: object,
) -> dict[str, object]

Update a single cell in a table.

Returns:

Type	Description
`dict[str, object]`	Dict with `ok` and `old_value` on success, or `ok` and `error`.

build_submissions_query ¶

build_submissions_query(
    *, where: str = "", order: str = "", limit: int = 0
) -> str

Build the shared submissions dataset query.

get_enriched_rows ¶

get_enriched_rows(
    sdb: Database, table: str
) -> list[dict[str, object]]

Return all rows from a table using enriched query if available.

run_query ¶

run_query(query: str) -> list[dict[str, object]]

Execute arbitrary SQL and return results as list of dicts.

sql ¶

sql(query: str) -> QueryResult

Execute SQL and return a QueryResult with .columns and .fetchall().

canvas_apply ¶

canvas_apply(
    conn: Database, pending: _PendingChanges
) -> dict[str, object]

Push pending changes to Canvas and clear synced rows on success.

canvas_preview ¶

canvas_preview(
    conn: Database, pending: _PendingChanges
) -> dict[str, object]

Compare pending changes against live Canvas state.

get_pending_changes ¶

get_pending_changes(
    sdb: Database | None = None,
) -> dict[str, dict[str, dict[str, dict[str, object]]]]

Compute pending changes by diffing main tables against synced shadows.

Returns the same PendingChanges structure used by the viewer: {table: {pk_key: {column: {"baseline": ..., "current": ...}}}}.

get_pull_blocking_tables ¶

get_pull_blocking_tables(
    sdb: Database | None = None,
) -> list[str]

Return Canvas-managed working tables with pending local edits.

mark_synced_assignments ¶

mark_synced_assignments(
    sdb: Database, canvas_ids: list[int]
) -> None

Update synced shadow for specific assignments after push.

mark_synced_grades ¶

mark_synced_grades(
    sdb: Database, keys: list[tuple[int, int]]
) -> None

Update synced shadow for specific grade rows after push.

preview_assignments ¶

preview_assignments(
    conn: Database,
    table_changes: _TableChanges,
    client: CanvasClient,
) -> list[dict[str, object]]

Preview pending assignment changes against live Canvas state.

preview_grades ¶

preview_grades(
    conn: Database,
    table_changes: _TableChanges,
    client: CanvasClient,
) -> list[dict[str, object]]

Preview pending grade changes against live Canvas submissions.

pull_block_reason ¶

pull_block_reason(
    sdb: Database | None = None,
) -> str | None

Describe why pull is blocked, or return None when pull is allowed.

snapshot_canvas_synced ¶

snapshot_canvas_synced(sdb: Database | None = None) -> None

Copy current canvas tables into synced shadow tables.

Called after a successful pull to record what Canvas has.

build_canvas_gradebook_matrix ¶

build_canvas_gradebook_matrix(
    conn: Database,
) -> tuple[list[str], list[list[str]]]

Build a gradebook matrix for CLI/report output.

load_canvas_gradebook_data ¶

load_canvas_gradebook_data(
    conn: Database,
) -> CanvasGradebookData

Load shared Canvas gradebook data for both CLI and viewer.

catalog ¶

Shared catalog metadata for user-facing cass data workflows.

TableCapability `dataclass` ¶

TableCapability(
    editable: bool = False,
    pushable: bool = False,
    pull_guarded: bool = False,
    viewer_rank: int = 999,
    editable_columns: frozenset[str] | None = None,
)

Shared table behavior metadata for CLI and viewer workflows.

get_table_capability ¶

get_table_capability(table: str) -> TableCapability

Return shared behavior metadata for a table.

core ¶

SQLite database for cass — connection, schema, meta, and CRUD.

Schema v16: Canvas source tables (canvas_*) plus Canvas grade tables for manual push workflows. Synced shadow tables (_canvas_assignments_synced, _canvas_grades_synced) provide persistent change tracking between local edits and Canvas state.

db_path ¶

db_path(root: Path | None = None) -> str

Return the SQLite file path.

connect_db ¶

connect_db(
    root: Path | None = None,
) -> sqlite_utils.Database

Open a lightweight connection to an existing database.

Unlike open_db, this skips schema init and reconciliation — use it when you know the DB already exists (e.g. in a background thread while the viewer's main connection is alive).

open_db ¶

open_db(root: Path | None = None) -> sqlite_utils.Database

Open a database for the resolved project root and ensure schema exists.

get_db ¶

get_db(root: Path | None = None) -> sqlite_utils.Database

Return the shared Database, creating it on first call.

editable_canvas_tables ¶

editable_canvas_tables() -> set[str]

Return the current editable Canvas-managed working tables.

save_meta ¶

save_meta(key: str, value: str) -> None

Store a key-value pair in the meta table.

get_meta ¶

get_meta(
    key: str, sdb: Database | None = None
) -> str | None

Retrieve a value from the meta table, or None if not found.

save_canvas_students ¶

save_canvas_students(students: list[CanvasStudent]) -> int

Upsert Canvas students into the source table.

load_canvas_student_ids ¶

load_canvas_student_ids() -> set[int]

Return every canvas_id in the Canvas roster.

save_canvas_assignments ¶

save_canvas_assignments(
    assignments: list[CanvasAssignment],
) -> int

Save Canvas assignments from domain models.

load_canvas_assignment_ids ¶

load_canvas_assignment_ids() -> list[int]

Return every Canvas assignment id, ascending.

save_canvas_submissions ¶

save_canvas_submissions(
    subs: list[CanvasSubmission],
) -> int

Upsert Canvas submissions.

upsert_canvas_grade ¶

upsert_canvas_grade(
    sdb: Database,
    canvas_user_id: int,
    canvas_assignment_id: int,
    posted_grade: str,
) -> str

Insert or update a single canvas grade, returning the previous value.

Returns:

Type	Description
`str`	Previous posted_grade (empty string if row didn't exist).

get_assignment_groups ¶

get_assignment_groups(sdb: Database) -> list[str]

Return distinct non-empty assignment groups, sorted alphabetically.

save_canvas_grades ¶

save_canvas_grades(grades: list[CanvasGrade]) -> int

Upsert Canvas grades.

load_canvas_grades ¶

load_canvas_grades(
    canvas_assignment_id: int | None = None,
) -> list[CanvasGrade]

Load Canvas grades.

ibis_adapter ¶

Generic DB adapter via ibis-framework — DuckDB and SQLite backends.

connect_file ¶

connect_file(filepath: Path) -> BaseBackend

Connect to a DuckDB or SQLite file via ibis.

Raises:

Type	Description
`SystemExit`	If the file does not exist or has an unsupported suffix.

list_tables ¶

list_tables(con: BaseBackend) -> list[dict[str, str]]

List all tables in the connected database.

get_schema ¶

get_schema(
    con: BaseBackend, table: str
) -> list[tuple[str, str]]

Return column names and type strings for a table.

get_primary_keys ¶

get_primary_keys(con: BaseBackend, table: str) -> list[str]

Detect primary key columns via PRAGMA table_info.

Works for both DuckDB (pk flag is bool) and SQLite (pk flag is int > 0). Falls back to empty list if detection fails.

get_rows ¶

get_rows(
    con: BaseBackend, table: str, *, limit: int = 10000
) -> list[dict[str, Any]]

Fetch rows from a table as sanitized list of dicts.

get_row_count ¶

get_row_count(con: BaseBackend, table: str) -> int

Return the number of rows in a table.

update_cell ¶

update_cell(
    con: BaseBackend,
    table: str,
    pk_col: str,
    pk_val: Any,
    column: str,
    value: Any,
) -> dict[str, Any]

Update a single cell value and return status dict.

get_categorical_columns ¶

get_categorical_columns(
    con: BaseBackend, table: str, *, max_distinct: int = 10
) -> dict[str, list[str]]

Detect string columns with few distinct values and return their options.

Parameters:

Name	Type	Description	Default
`con`	`BaseBackend`	ibis connection.	required
`table`	`str`	Table name.	required
`max_distinct`	`int`	Maximum distinct values to qualify as categorical.	`10`

Returns:

Type	Description
`dict[str, list[str]]`	Mapping of column name to sorted list of distinct values.

run_sql ¶

run_sql(
    con: BaseBackend, sql: str
) -> tuple[list[str], list[dict[str, Any]]]

Execute raw SQL and return (column_names, list_of_row_dicts).

backend_name ¶

backend_name(con: BaseBackend) -> str

Return the backend name: 'duckdb' or 'sqlite'.

introspection ¶

Viewer DB introspection — table metadata, cell editing, and revert.

Functions used by the viewer (and tests) to inspect table structure, update individual cells, and revert pending changes.

get_tables ¶

get_tables(sdb: Database) -> list[dict[str, str]]

Return list of non-excluded tables with their type.

get_primary_keys ¶

get_primary_keys(sdb: Database, table: str) -> list[str]

Return primary key column names for a table.

get_column_names ¶

get_column_names(sdb: Database, table: str) -> list[str]

Return column names for a table.

is_editable ¶

is_editable(sdb: Database, table: str) -> bool

Check if a table is editable (has PKs, is a real table, not read-only).

update_cell ¶

update_cell(
    sdb: Database,
    table: str,
    pk: dict[str, object],
    column: str,
    value: object,
) -> dict[str, object]

Update a single cell in a table.

Returns:

Type	Description
`dict[str, object]`	Dict with `ok` and `old_value` on success, or `ok` and `error`.

revert_changes ¶

revert_changes(
    sdb: Database,
    pending: dict[
        str, dict[str, dict[str, dict[str, object]]]
    ],
) -> int

Revert all pending changes in the DB by restoring baseline values.

Returns:

Type	Description
`int`	Number of field changes reverted.

queries ¶

Enriched queries and shared query dataset helpers.

QueryResult ¶

QueryResult(
    columns: list[str], rows: list[tuple[object, ...]]
)

Thin wrapper around a sqlite3 cursor for Rich table rendering.

build_submissions_query ¶

build_submissions_query(
    *, where: str = "", order: str = "", limit: int = 0
) -> str

Build the shared submissions dataset query.

get_enriched_rows ¶

get_enriched_rows(
    sdb: Database, table: str
) -> list[dict[str, object]]

Return all rows from a table using enriched query if available.

sql ¶

sql(query: str) -> QueryResult

Execute SQL and return a QueryResult with .columns and .fetchall().

run_query ¶

run_query(query: str) -> list[dict[str, object]]

Execute arbitrary SQL and return results as list of dicts.

schema ¶

Domain models for cass — Canvas students, assignments, submissions, grades.

CanvasStudent ¶

Bases: Struct

A Canvas LMS student record.

from_api `classmethod` ¶

from_api(
    s: CanvasStudentResponse, *, sis_section_id: str = ""
) -> CanvasStudent

Convert a Canvas API student response to a domain model.

CanvasAssignment ¶

Bases: Struct

A Canvas LMS assignment.

from_api `classmethod` ¶

from_api(
    a: CanvasAssignmentResponse, *, group_name: str = ""
) -> CanvasAssignment

Convert a Canvas API assignment response to a domain model.

CanvasSubmission ¶

Bases: Struct

A Canvas LMS submission record.

CanvasGrade ¶

Bases: Struct

A grade ready for Canvas API push.

sync ¶

Canvas sync state and shared preview/apply workflows.

snapshot_canvas_synced ¶

snapshot_canvas_synced(sdb: Database | None = None) -> None

Copy current canvas tables into synced shadow tables.

Called after a successful pull to record what Canvas has.

mark_synced_assignments ¶

mark_synced_assignments(
    sdb: Database, canvas_ids: list[int]
) -> None

Update synced shadow for specific assignments after push.

mark_synced_grades ¶

mark_synced_grades(
    sdb: Database, keys: list[tuple[int, int]]
) -> None

Update synced shadow for specific grade rows after push.

get_pending_changes ¶

get_pending_changes(
    sdb: Database | None = None,
) -> dict[str, dict[str, dict[str, dict[str, object]]]]

Compute pending changes by diffing main tables against synced shadows.

Returns the same PendingChanges structure used by the viewer: {table: {pk_key: {column: {"baseline": ..., "current": ...}}}}.

get_pull_blocking_tables ¶

get_pull_blocking_tables(
    sdb: Database | None = None,
) -> list[str]

Return Canvas-managed working tables with pending local edits.

pull_block_reason ¶

pull_block_reason(
    sdb: Database | None = None,
) -> str | None

Describe why pull is blocked, or return None when pull is allowed.

preview_assignments ¶

preview_assignments(
    conn: Database,
    table_changes: _TableChanges,
    client: CanvasClient,
) -> list[dict[str, object]]

Preview pending assignment changes against live Canvas state.

preview_grades ¶

preview_grades(
    conn: Database,
    table_changes: _TableChanges,
    client: CanvasClient,
) -> list[dict[str, object]]

Preview pending grade changes against live Canvas submissions.

canvas_preview ¶

canvas_preview(
    conn: Database, pending: _PendingChanges
) -> dict[str, object]

Compare pending changes against live Canvas state.

canvas_apply ¶

canvas_apply(
    conn: Database, pending: _PendingChanges
) -> dict[str, object]

Push pending changes to Canvas and clear synced rows on success.

views ¶

Shared view builders for CLI and viewer parity.

load_canvas_gradebook_data ¶

load_canvas_gradebook_data(
    conn: Database,
) -> CanvasGradebookData

Load shared Canvas gradebook data for both CLI and viewer.

build_canvas_gradebook_matrix ¶

build_canvas_gradebook_matrix(
    conn: Database,
) -> tuple[list[str], list[list[str]]]

Build a gradebook matrix for CLI/report output.

cass.db¶

cass.db ¶

TableCapability dataclass ¶

QueryResult ¶

CanvasAssignment ¶

from_api classmethod ¶

CanvasGrade ¶

CanvasStudent ¶

from_api classmethod ¶

CanvasSubmission ¶

get_table_capability ¶

connect_db ¶

db_path ¶

editable_canvas_tables ¶

get_assignment_groups ¶

get_db ¶

get_meta ¶

load_canvas_assignment_ids ¶

load_canvas_grades ¶

load_canvas_student_ids ¶

open_db ¶

save_canvas_assignments ¶

save_canvas_grades ¶

save_canvas_students ¶

save_canvas_submissions ¶

save_meta ¶

upsert_canvas_grade ¶

get_column_names ¶

get_primary_keys ¶

get_tables ¶

is_editable ¶

revert_changes ¶

update_cell ¶

build_submissions_query ¶

get_enriched_rows ¶

run_query ¶

sql ¶

canvas_apply ¶

canvas_preview ¶

get_pending_changes ¶

get_pull_blocking_tables ¶

mark_synced_assignments ¶

mark_synced_grades ¶

preview_assignments ¶

preview_grades ¶

pull_block_reason ¶

snapshot_canvas_synced ¶

build_canvas_gradebook_matrix ¶

load_canvas_gradebook_data ¶

catalog ¶

TableCapability dataclass ¶

get_table_capability ¶

core ¶

db_path ¶

connect_db ¶

open_db ¶

get_db ¶

editable_canvas_tables ¶

save_meta ¶

get_meta ¶

save_canvas_students ¶

load_canvas_student_ids ¶

save_canvas_assignments ¶

load_canvas_assignment_ids ¶

save_canvas_submissions ¶

upsert_canvas_grade ¶

get_assignment_groups ¶

save_canvas_grades ¶

load_canvas_grades ¶

ibis_adapter ¶

connect_file ¶

list_tables ¶

get_schema ¶

get_primary_keys ¶

get_rows ¶

get_row_count ¶

update_cell ¶

get_categorical_columns ¶

run_sql ¶

backend_name ¶

TableCapability `dataclass` ¶

from_api `classmethod` ¶

from_api `classmethod` ¶

TableCapability `dataclass` ¶

from_api `classmethod` ¶

from_api `classmethod` ¶