API reference¶

def set_filters(self, specs: list[FilterSpec]) -> None:
    """Show chips for a list of filters (e.g. when restoring state)."""
    self._clear_all(emit_signal=False)
    for spec in specs:
        self._add_chip(spec)

def add_condition(self, field: str, op: str, value: float, *, notify: bool = True) -> None:
    """Add (or replace) a condition: the same field+operator is an edit, not another chip."""
    if field not in self._labels or op not in dict(OPERATORS):
        return
    self._drop((field, op))
    self._conditions[(field, op)] = float(value)
    symbol = dict(OPERATORS)[op]
    chip = _Chip(field, op, f"{self._labels[field]} {symbol} {float(value):g}", self)
    chip.removed.connect(self._on_removed)
    self._chips[(field, op)] = chip
    self._row.insertWidget(self._row.count() - 1, chip)  # before the stretch
    self._empty.setVisible(False)
    if notify:
        self.changed.emit()

def set_conditions(self, conditions: Iterable[Condition]) -> None:
    """Whole state at once (restoring a saved view): a single signal."""
    for key in list(self._chips):
        self._drop(key)
    for field, op, value in conditions:
        self.add_condition(field, op, value, notify=False)
    self.changed.emit()

def set_external_clause(self, key: str, clause) -> None:
    """Overlay a raw SQLAlchemy clause under `key` (None removes it).

    Escape hatch for what FilterSpec cannot express - subqueries
    (`~exists(...)`), OR across fields, functions. It composes with AND just
    like `set_external_filters`, and because it is an already-built expression
    the builder does not interpret it: making sure it references the right
    model is the caller's responsibility.
    """
    if clause is None:
        self._external_clauses.pop(key, None)
    else:
        self._external_clauses[key] = clause
    self._page = 1

def set_search(self, text: str, fields: list[str]) -> None:
    """Enable an ILIKE search ORed across the given fields."""
    self._search_text   = text
    self._search_fields = fields
    self._page = 1

def remove_search(self) -> None:
    """Disable the quick search and show everything again."""
    self._search_text   = ""
    self._search_fields = []
    self._page = 1

def all_ids(self) -> list[int]:
    """PKs of every row matching the current filters/search (no pagination).

    Lets a caller resolve "everything the table currently shows" into an id set
    without materializing rows. Honors filters + quick search, ignores paging.
    """
    pk = self._get_pk_attr()
    if pk is None:
        return []
    stmt = self._build_base_statement().with_only_columns(pk)
    with self._db.get_session() as session:
        return [int(value) for value in session.exec(stmt).all() if value is not None]

def count(self) -> int:
    """Row total under the current filters/search, without fetching any row.

    For hosts that want to keep the counter live (e.g. while a job inserts
    records) without reloading the visible rows.
    """
    with self._db.get_session() as session:
        return int(session.exec(self._build_count_statement()).one())  # type: ignore[arg-type]

def cursor_of(self, row: dict) -> tuple[Any, Any] | None:
    """Keyset cursor `(sort_value, pk)` for an already-loaded row."""
    raw = row.get("__raw__")
    pk = self._get_pk_attr()
    if raw is None or pk is None:
        return None
    pk_value = getattr(raw, pk.key, None)
    if not self._sorts:
        return (None, pk_value)
    sort = self._sorts[0]
    col_def = self._config.column_by_field(sort.field)
    key = col_def.display_key if col_def is not None else sort.field
    return ((row.get("__cursor_values__") or {}).get(key, row.get(key)), pk_value)

def fetch_after(self, cursor: tuple[Any, Any] | None) -> tuple[list[dict], int]:
    """Next page by keyset: `WHERE (sort, pk) > cursor`, with no OFFSET.

    OFFSET makes the engine walk and discard the previous N rows, so page 500
    costs 500 times the first one. The cursor does not: it is always an index
    seek. The cursor is composite because the visible order is rarely unique
    (two identical scores).
    """
    stmt = self._build_base_statement()
    clause = self._keyset_clause(cursor)
    if clause is not None:
        stmt = stmt.where(clause)
    with self._db.get_session() as session:
        total: int = session.exec(self._build_count_statement()).one()  # type: ignore[arg-type]
        rows = [self._row_to_dict(r) for r in session.exec(stmt.limit(self._page_size)).all()]
    return rows, total

def fetch_window_after(self, cursor: tuple[Any, Any] | None) -> tuple[list[dict], bool]:
    """The page after the cursor, without ``COUNT(*)``.

    Infinite scroll only needs to know whether another page exists. Asking for
    one extra row answers that without re-counting a potentially huge table on
    every move.
    """
    stmt = self._build_base_statement()
    clause = self._keyset_clause(cursor)
    if clause is not None:
        stmt = stmt.where(clause)
    with self._db.get_session() as session:
        raw = session.exec(stmt.limit(self._page_size + 1)).all()
    has_more = len(raw) > self._page_size
    return [self._row_to_dict(row) for row in raw[:self._page_size]], has_more

def fetch_window_at(self, page_index: int) -> tuple[list[dict], bool]:
    """Direct load for a scrollbar jump to a page that has not been walked yet.

    The normal path is still keyset via :meth:`fetch_window_after`. OFFSET only
    shows up when the user drags the scrollbar to a position with no known
    cursor.
    """
    offset = max(0, int(page_index)) * self._page_size
    stmt = self._build_base_statement().offset(offset).limit(self._page_size + 1)
    with self._db.get_session() as session:
        raw = session.exec(stmt).all()
    has_more = len(raw) > self._page_size
    return [self._row_to_dict(row) for row in raw[:self._page_size]], has_more

def fetch(self) -> tuple[list[dict], int]:
    """Run the query and return (rows_as_dicts, total_count)."""
    stmt      = self._build_base_statement()
    count_stmt = self._build_count_statement()
    with self._db.get_session() as session:
        total: int = session.exec(count_stmt).one()  # type: ignore[arg-type]

        offset = (self._page - 1) * self._page_size
        stmt   = stmt.offset(offset).limit(self._page_size)

        raw_rows = session.exec(stmt).all()
        rows     = [self._row_to_dict(r) for r in raw_rows]
    return rows, total

def to_query_spec(self) -> dict:
    """The "filtered" selection as QuerySpec kwargs, ready to hand to a job.

    Returns a dict rather than a QuerySpec so the table does not take ms_flow as
    a dependency; the host does `QuerySpec(**builder.to_query_spec())`. This is
    what avoids materialising a list of ids: the job receives the query, not its
    result.

    Raises when any part of the current state is not expressible (expression
    columns, LIKE/ILIKE filters over joins): staying quiet and sending a wider
    query would be worse.
    """
    model = self._config.model_class
    table = getattr(getattr(model, "__table__", None), "name", "") or model.__name__.lower()

    filters: dict[str, Any] = {}
    for f in self._all_filters():
        col_def = self._config.column_by_field(f.field)
        if col_def is not None and col_def.expr is not None:
            raise ValueError(f"Column '{f.field}' filters on a SQL expression; QuerySpec cannot carry it.")
        if "." in f.field:
            raise ValueError(f"Filter on joined field '{f.field}' has no QuerySpec equivalent.")
        if f.op in (FilterOperator.LIKE, FilterOperator.ILIKE):
            filters[f"{f.field}__contains"] = str(f.value).strip("%")
            continue
        suffix = self._SPEC_OPS.get(f.op)
        if suffix is None:
            raise ValueError(f"Filter operator {f.op} has no QuerySpec equivalent.")
        filters[f"{f.field}{suffix}"] = f.value

    if self._external_clauses:
        raise ValueError("Raw SQLAlchemy clauses set by the host cannot be exported to QuerySpec.")
    if self._search_text and self._search_fields:
        # Quick search is an OR across columns; QuerySpec filters are ANDed.
        raise ValueError("Quick search (OR across columns) has no QuerySpec equivalent.")

    order = tuple(
        f"-{s.field}" if s.descending else s.field
        for s in self._sorts
        if "." not in s.field
    )
    return {"table": table, "filters": filters, "order": order}

def get_session(self) -> Session:
    """Return a fresh SQLModel session for a query cycle."""

def refresh_counts(self) -> bool:
    """Refresh only the record counter (COUNT), without reloading rows.

    Meant for running jobs: the user watches the total grow and infinite scroll
    knows there are more pages, but the visible rows do not shift under the
    cursor. Returns False when the table is empty (nothing to preserve: the
    host should do a plain refresh()).
    """
    if self._model.loaded_count == 0:
        return False
    total = self._builder.count()
    if self._is_infinite_mode():
        self._infinite_total_hint = total
        self._infinite_total_exact = True
        self._infinite_has_more = self._infinite_loaded_end < total
        tail = self._infinite_pages.get(self._infinite_frontier_page)
        if tail is not None and len(tail) < self._builder.page_size and self._infinite_has_more:
            # The last page came back short and rows were added since: forget it, so the
            # autofill set_total() triggers fetches it again once it is in view. Kept, its
            # rows past the old end would stay blank forever.
            del self._infinite_pages[self._infinite_frontier_page]
            # The next page's cursor is the short page's last row: wrong now, so it is
            # found by offset instead.
            self._infinite_page_cursors.pop(self._infinite_frontier_page + 1, None)
    self._model.set_total(total)
    return True

def ensure_viewport_filled(self, *, force: bool = False) -> bool:
    """One tick of a job's start-up: reload rows only once there are enough to
    fill the visible viewport (or once the host decides it has waited long
    enough, force=True).

    Returns True when the viewport is full (or there is nothing else to fetch):
    the host then switches to "counters only" mode. False = still waiting, call
    again.
    """
    loaded = self._model.loaded_count
    capacity = self._viewport_row_capacity()
    if loaded >= capacity:
        return True
    total = self._builder.count()
    if not force and total < capacity:
        return False  # not enough molecules yet to fill the view
    if total <= loaded:
        return force
    self.refresh()
    return force or self._model.loaded_count >= capacity

def patch_loaded_rows(self, key: str, patches: dict[Any, dict[str, Any]]) -> bool:
    """Patch visible cached rows by a raw-object/key value; no model reset or reselection."""
    changed = False
    for row in self._model.loaded_rows:
        data = self._model.get_row_data(row) or {}
        raw = data.get("__raw__")
        identity = data.get(key, getattr(raw, key, None))
        values = patches.get(identity)
        if values is not None:
            changed = self._model.patch_loaded_row(row, values) or changed
    return changed

def open_table_menu(self, anchor: QWidget | None = None) -> None:
    """Pop up the full menu for external hosts (a ribbon, for instance).

    The embedded toolbar exposes these options as actions of its own.
    ``anchor`` lets another host open the same collection under one of its own
    controls.
    """
    menu = QMenu(self)
    columns_action = menu.addAction("Columns", self._open_column_config)
    columns_action.setIcon(self._toolbar_icon("columns.svg"))
    menu.addMenu(self._create_export_menu(menu))
    if self._toolbar is not None:
        menu.addMenu(self._create_toolbar_style_menu(menu))

    if anchor is not None:
        menu.exec(anchor.mapToGlobal(anchor.rect().bottomLeft()))
    else:
        from PySide6.QtGui import QCursor
        menu.exec(QCursor.pos())

def set_external_filters(self, key: str, filters: list[FilterSpec]) -> None:
    """Set (or replace) an external filter overlay under `key` and reload.

    It composes with the user's interactive filters using AND, without touching
    them; passing an empty list is the same as `clear_external_filters(key)`.
    """
    self._builder.set_external_filters(key, filters)
    self.refresh()

def set_external_clause(self, key: str, clause) -> None:
    """Set (or, with None, drop) a raw SQLAlchemy clause under `key` and reload.

    For conditions FilterSpec cannot express - typically a `~exists(...)`
    against another table.
    """
    self._builder.set_external_clause(key, clause)
    self.refresh()

def set_empty_state(self, message: str | None = None, *, show_action: bool | None = None):
    """Change the empty state at runtime: `None` restores what the config declares.

    A host that filters the table from outside knows something the table does
    not: "there is no data" and "the filter lets nothing through" are different
    states, and the action that fills the table (import, create) does not fix
    the second one.
    """
    self._empty_label.setText(message or self._config.empty_message)
    button = self._empty_widget.findChild(QPushButton, "EmptyActionButton")
    if button is not None:
        button.setVisible(True if show_action is None else bool(show_action))

def set_toolbar_button_style(self, style: Qt.ToolButtonStyle) -> None:
    """Icons / text / both for the toolbar actions."""
    if self._toolbar is not None:
        self._toolbar.setToolButtonStyle(style)

def set_toolbar_actions(self, key: str, actions: list[ToolbarAction]) -> None:
    """Put a host's commands on this table's toolbar while it needs them.

    Same shape as `set_external_filters`: keyed by the host (a tool id), replacing
    whatever that key had, and an empty list is `clear_toolbar_actions(key)`. A tool that
    narrows a table can now also act on it, and gives both back on the way out.
    """
    self.clear_toolbar_actions(key)
    if not actions:
        return
    if self._toolbar is None:
        # Config asked for no toolbar; a runtime action still has to be reachable.
        self.layout().insertWidget(0, self._build_toolbar())
    added = [self._make_action(a) for a in actions]
    for act in added:
        act.setVisible(not self._actions_collapsed)
        self._toolbar.insertAction(self._zone_anchor, act)
    self._keyed_actions[key] = added

def set_config_actions_visible(self, visible: bool) -> None:
    """Show/hide the actions the config declares (toolbar_left/center/right).

    The built-in controls (Columns, Export, Reload, Settings) stay: they describe the
    table. The config's own actions describe the *catalog* — a table lent to a tool
    should not still be offering "Import…".
    """
    self._config_actions_visible = bool(visible)
    self._apply_collapsed_state()

def view_state(self) -> dict:
    """Current user-facing table prefs: visible column fields + active sort.
    Suitable for JSON/TOML persistence by the host app."""
    return {
        "columns": [c.field for c in self._config.columns if c.visible],
        "sort": [{"field": s.field, "descending": bool(s.descending)} for s in self._builder.active_sorts],
    }

def apply_view_state(self, state: dict | None) -> None:
    """Apply a previously saved view_state(). Unknown fields are ignored;
    an empty/missing 'columns' leaves the config-defined visibility intact."""
    if not state:
        return
    columns = state.get("columns")
    if columns:
        wanted = set(columns)
        known = {c.field for c in self._config.columns}
        # Only apply if it references real columns; keep at least one visible.
        if wanted & known:
            for c in self._config.columns:
                c.visible = c.field in wanted
    sort = state.get("sort")
    if sort:
        self._builder.set_sort(
            [SortSpec(s["field"], descending=bool(s.get("descending"))) for s in sort if s.get("field")]
        )
    self._install_delegates()
    self._sync_header()
    self.refresh()

def refresh(self, reset_page: bool = True) -> None:
    """Reload data from the DB with the current filter/sort/page state."""
    if self._is_infinite_mode():
        self._builder.set_page(1)
        rows, _has_more = self._builder.fetch_window_after(None)
        self._infinite_pages = {0: rows}
        self._infinite_page_cursors = {0: None}
        if rows:
            self._infinite_page_cursors[1] = self._builder.cursor_of(rows[-1])
        # ponytail: one COUNT per refresh (not per scrolled page). "100+ records" is not
        # an answer to "how many are there"; move this off-thread if it ever profiles hot.
        total = self._builder.count()
        self._infinite_frontier_page = 0
        self._infinite_loaded_end = len(rows)
        self._infinite_has_more = bool(rows) and len(rows) < total
        self._infinite_total_hint = total
        self._infinite_total_exact = True
        self._model.load_data(
            rows,
            total,
            1,
            window_start=0,
            total_is_exact=True,
        )
        return
    if reset_page:
        self._builder.set_page(1)

    rows, total = self._builder.fetch()
    self._model.load_data(rows, total, self._builder.current_page)

def load_next_page(self) -> None:
    """Load the next page and drop old pages that fell outside the window."""
    if not self._is_infinite_mode() or self._loading_more:
        return
    if not self._infinite_has_more:
        return
    self._load_infinite_page(self._infinite_frontier_page + 1)

def set_db(self, db: SessionProvider) -> None:
    """Replace the session provider (useful on project switch or reconnect)."""
    self._db = db
    self._builder = QueryBuilder(db, self._config)
    self._update_control_summaries()
    self.refresh()

def repaint_rows(self) -> None:
    """Repaint the visible rows without re-querying (host state changed, not the data)."""
    self._table.viewport().update()

def get_selected_objects(self) -> list[Any]:
    """Return the list of selected SQLModel objects."""
    indexes = self._table.selectionModel().selectedRows()
    objs = []
    for idx in indexes:
        raw = self._model.data(idx, RAW_OBJECT_ROLE)
        if raw is not None:
            objs.append(raw)
    return objs

def get_selected_object(self) -> Optional[Any]:
    """Return the first selected object, or None."""
    objs = self.get_selected_objects()
    return objs[0] if objs else None

def all_filtered_ids(self) -> list[int]:
    """PKs of every row matching the current filters/search (across all pages)."""
    return self._builder.all_ids()

def set_filter(self, spec: FilterSpec) -> None:
    """Apply a filter programmatically."""
    self._builder.add_filter(spec)
    self._sync_header()
    self._update_control_summaries()
    self.refresh()

def set_filters(self, filters: list[FilterSpec]) -> None:
    """Replace the whole scope and reload exactly once.

    This is the right operation for external controls that change receptor,
    ligand, protocol and metrics as a single state transition.
    """
    self._builder.set_filters(list(filters))
    self._sync_header()
    self._update_control_summaries()
    self.refresh()

def select_first_row(self) -> bool:
    """Select the first materialised row; never the sentinel row."""
    if self._model.loaded_count == 0:
        return False
    self._table.selectRow(self._model.window_start)
    return True