devboard ¤

def force_refresh_item(self, item: Any, /) -> None:
    """Prepare one item for a forced scan."""
    self.refresh_item(item)

def matches_filter(self, row: Row[Any], value: str, /) -> bool:
    """Match filter text against displayed cells, ignoring case.

    Override this method to match source items instead. For example, a
    backlog board can compare `row.item.repository` with `value`.
    An empty filter shows every row.
    """
    return not value or any(value.casefold() in str(cell).casefold() for cell in row.data)

def refresh_item(self, item: Any, /) -> None:
    """Prepare one item for a normal scan."""

def check(self) -> None:
    """Uncheck the checkbox."""
    self.checked = True

def toggle(self) -> bool:
    """Toggle the checkbox."""
    self.checked = not self.checked
    return self.checked

def uncheck(self) -> None:
    """Uncheck the checkbox."""
    self.checked = False

def action_toggle_collapse(self) -> None:
    """Collapse or expand the column."""
    self.is_collapsed = not self.is_collapsed
    self.screen.set_focus(self if self.is_collapsed else self.table)

def action_toggle_maximize(self) -> None:
    """Maximize the column or restore the layout from before it was maximized."""
    if self._layout_before_maximize is None:
        columns = list(self.screen.query(Column))
        self._layout_before_maximize = [(column, column.is_collapsed) for column in columns]
        for column in columns:
            if column is self:
                column._expand()
            else:
                column._collapse()
        self.screen.set_focus(self.table)
        return

    layout = self._layout_before_maximize
    self._layout_before_maximize = None
    for column, was_collapsed in layout:
        if was_collapsed:
            column._collapse()
        else:
            column._expand()
    self.screen.set_focus(self if self.is_collapsed else self.table)

def compose(self) -> ComposeResult:
    """Compose column widgets."""
    with Horizontal(classes="column-header"):
        yield Static("▶ " + self.TITLE, classes="column-title")
        yield Static("0", classes="column-count", markup=False)
    yield DataTable(id="table")

def deserialize_cell(self, value: Any) -> Any:
    """Restore a cell value loaded from the cache."""
    return value

def filter_rows(self, predicate: Callable[[Row[_ItemT]], bool] | None) -> None:
    """Show matching rows, or clear this column's filter with `None`.

    The predicate receives each row, including its source item and display
    cells. It also applies after refreshes. Call this method on the UI thread.
    """
    was_empty = not self.table.row_count
    self.table.filter_rows((lambda row: predicate(cast("Row[_ItemT]", row))) if predicate is not None else None)
    if was_empty and self.table.row_count:
        self._expand()
    self._finalize()

def item_key(self, item: _ItemT, /) -> Hashable:
    """Return the identity used to share and cache an item.

    Keys must be hashable and unique across the board.
    Objects can expose a `devboard_key` attribute to provide a stable
    identity. Hashable objects otherwise use their own identity. Unhashable
    objects use their process-local identity and should override this method
    if their rows need to be restored from the on-disk cache.
    """
    key = getattr(item, "devboard_key", item)
    try:
        hash(key)
    except TypeError:
        return id(item)
    return cast("Hashable", key)

def list_items(self) -> Iterable[_ItemT]:
    """List the items to scan for this column."""
    return ()

def modal(self, text: RenderableType) -> None:
    """Ask the UI thread to show text or a Rich renderable in a modal."""
    self.app.call_later(self._push_modal, text)

def notify_error(self, message: str, timeout: float = 3.0) -> None:
    """Notify error."""
    self.app.notify(f"[b red]ERROR[/]  {message}", severity="error", timeout=timeout)

def notify_info(self, message: str, timeout: float = 3.0) -> None:
    """Notify information."""
    self.app.notify(f"[b blue]INFO[/]  {message}", severity="information", timeout=timeout)

def notify_success(self, message: str, timeout: float = 3.0) -> None:
    """Notify success."""
    self.app.notify(f"[b green]SUCCESS[/]  {message}", severity="information", timeout=timeout)

def notify_warning(self, message: str, timeout: float = 3.0) -> None:
    """Notify warning."""
    self.app.notify(f"[b yellow]WARNING[/]  {message}", severity="warning", timeout=timeout)

def populate_rows(self, item: _ItemT, /) -> list[tuple[Any, ...]]:  # noqa: ARG002
    """Build table rows for an item."""
    return []

def report_progress(self, description: str | None = None) -> None:
    """Show progress on the left of the footer.

    Devboard automatically clears progress associated with a worker when that worker finishes or is cancelled.

    Parameters:
        description: Progress text. Omit the text to clear the current progress.
    """
    report_progress = getattr(self.app, "_report_progress", None)
    if report_progress is not None:
        report_progress(self, description)

def serialize_cell(self, value: Any) -> Any:
    """Convert a cell value to data that the cache can store."""
    return value

def update(self) -> None:
    """Update the column (ask the app to recompute its data)."""
    refresh_board = getattr(self.app, "refresh_board", None)
    if refresh_board is not None:
        self.app.call_later(refresh_board, [self])

def action_reverse_select(self) -> None:
    """Reverse selection."""
    for row in self.selectable_rows:
        row.toggle_select()
    self.force_refresh()

def action_toggle_select_all(self) -> None:
    """Toggle-select all rows."""
    rows = list(self.selectable_rows)
    if all(row.selected for row in rows):
        for row in rows:
            row.unselect()
    else:
        for row in rows:
            row.select()
    self.force_refresh()

def action_toggle_select_down(self) -> None:
    """Toggle selection down."""
    try:
        row = self.current_row
        next_row = row.next
    except CellDoesNotExist:
        pass
    else:
        next_row.toggle_select()
        self.move_cursor(row=next_row.index)
        self.force_refresh()

def action_toggle_select_row(self) -> None:
    """Toggle-select current row."""
    try:
        row = self.current_row
    except CellDoesNotExist:
        return
    row.toggle_select()
    self.force_refresh()

def action_toggle_select_up(self) -> None:
    """Toggle selection up."""
    try:
        row = self.current_row
        previous_row = row.previous
    except CellDoesNotExist:
        pass
    else:
        previous_row.toggle_select()
        self.move_cursor(row=previous_row.index)
        self.force_refresh()

def add_row(self, *cells: Any, height: int | None = 1, key: str | None = None, label: Any | None = None) -> RowKey:
    """Add a row and update the column's row count."""
    row_key = super().add_row(*cells, height=height, key=key, label=label)
    self.post_message(_TableRowsChanged())
    return row_key

def add_rows(self, rows: Iterable[Iterable]) -> list[RowKey]:
    """Add rows.

    Automatically insert a column with checkboxes in position 0.
    """
    return [self.add_row(*row) for row in rows]

def clear(self, columns: bool = True) -> DataTable[_ItemT]:  # noqa: FBT001,FBT002
    """Clear the table and update the column's row count."""
    super().clear(columns)
    self.post_message(_TableRowsChanged())
    return self

def filter_rows(self, predicate: Callable[[SelectableRow[_ItemT]], bool] | None) -> None:
    """Filter rows and update the column's visible row count."""
    super().filter_rows(predicate)
    self.post_message(_TableRowsChanged())

def force_refresh(self) -> None:
    """Force refresh table."""
    for row in self.selectable_rows:
        self.update_cell(row.key, "checkbox", row.checkbox)

def get_row(self, row_key: RowKey | str) -> list[Any]:
    """Return row cells, including cells retained while a row is hidden."""
    key = RowKey(row_key) if isinstance(row_key, str) else row_key
    if key in self._hidden_rows:
        return list(self._hidden_rows[key][0])
    return super().get_row(row_key)

def remove_row(self, row_key: RowKey | str) -> None:
    """Remove a row and report when the table becomes empty."""
    super().remove_row(row_key)
    self.post_message(_TableRowsChanged())
    if not self.row_count:
        self.post_message(_TableEmptied())

def sort(
    self,
    *columns: Any,
    key: Callable[[Any], Any] | None = None,
    reverse: bool = False,
) -> SelectableRowsDataTable[_ItemT]:
    """Sort all rows and retain the active filter."""
    column_indices = [self.get_column_index(column) for column in columns]

    def sort_key(row_key: RowKey) -> Any:
        cells = self.get_row(row_key)
        values = tuple(cells[index] for index in column_indices) if columns else tuple(cells)
        value = values[0] if len(columns) == 1 else values
        return key(value) if key is not None else value

    self._display_order = sorted(self._row_order, key=sort_key, reverse=reverse)
    self._sort_visible_rows()
    return self

def action_exit(self) -> None:
    """Exit application."""
    self._pending_item_refreshes.clear()
    self.workers.cancel_all()
    self.exit()

def action_filter_board(self) -> None:
    """Prompt for a filter to apply to all columns."""
    self._prompt_filter(list(self.query(Column)), "Filter board")

def action_filter_column(self) -> None:
    """Prompt for a filter to apply to the focused column."""
    if (column := self._focused_column()) is not None:
        self._prompt_filter([column], "Filter column")

def action_force_refresh_board(self) -> None:
    """Force-refresh all columns."""
    self.force_refresh_board()

def action_force_refresh_column(self) -> None:
    """Force-refresh the focused column."""
    if (column := self._focused_column()) is not None:
        self.force_refresh_board([column])

def action_force_refresh_item(self) -> None:
    """Force-refresh the item under the cursor in every column that lists it."""
    self._scan_current_item(force=True)

def action_refresh_board(self) -> None:
    """Refresh all columns."""
    self.refresh_board()

def action_refresh_column(self) -> None:
    """Refresh the focused column."""
    if (column := self._focused_column()) is not None:
        self.refresh_board([column])

def action_refresh_item(self) -> None:
    """Refresh the item under the cursor in every column that lists it."""
    self._scan_current_item(force=False)

def action_show_help(self) -> None:
    """Show help."""
    lines = ["## Main keys\n\n"]
    lines.extend(self._binding_specs_help(binding for _, binding in self._bindings))
    lines.append("\n\n## Selection\n\n")
    lines.extend(self._bindings_help(DataTable, search_up=True))
    lines.append("\n\n## Columns\n\n")
    lines.extend(self._bindings_help(Column))
    common_column_bindings = list(Binding.make_bindings(Column.BINDINGS))
    lines.append("\n\n## Current board\n\n")
    for column in self.query(Column):
        lines.append(f"\n\n### {column.__class__.TITLE}\n\n")
        bindings = (binding for _, binding in column._bindings if binding not in common_column_bindings)
        lines.extend(self._binding_specs_help(bindings))
    self.push_screen(Modal(text=Markdown("\n".join(lines))))

def action_show_help_panel(self) -> None:
    """Show widget help and grouped bindings for the focused column."""
    if not self.screen.query(HelpPanel):
        self.screen.mount(_KeysPanel())

def action_toggle_help_panel(self) -> None:
    """Show or hide Textual's keys and widget help panel."""
    if self.screen.query(HelpPanel):
        self.action_hide_help_panel()
    else:
        self.action_show_help_panel()

def compose(self) -> ComposeResult:
    """Compose the layout."""
    for column in self._columns:
        if isinstance(column, Column):
            yield column
        else:
            yield column()
    with Horizontal(id="status-bar"):
        yield self._progress
        yield Footer()

def filter_rows(self, value: str | None, columns: Iterable[Column] | None = None) -> None:
    """Filter specific columns, or all columns when none are specified.

    Use the board's `matches_filter()` hook to select rows. A nonempty value
    replaces each target column's filter. `None` or an empty string clears it.
    Filtering preserves hidden rows and does not scan items or write the cache.
    Call this method on the UI thread.
    """
    predicate = (lambda row: self.board.matches_filter(row, value)) if value else None
    for column in columns if columns is not None else self.query(Column):
        column.filter_rows(predicate)
        self._filter_values[column] = value or ""

def force_refresh_board(self, columns: Iterable[Column] | None = None) -> None:
    """Refresh columns after forcing their items to update external state."""
    self.scan(columns, force=True)

def get_system_commands(self, screen: Screen) -> Iterable[SystemCommand]:
    """Add refresh and filtering actions to the command palette."""
    yield from super().get_system_commands(screen)
    for action, (title, description) in {**_REFRESH_COMMANDS, **_FILTER_COMMANDS}.items():
        yield SystemCommand(title, description, getattr(self, f"action_{action}"))

def modal(self, text: RenderableType) -> None:
    """Ask the UI thread to show text or a Rich renderable in a modal."""
    self.app.call_later(self._push_modal, text)

def on_mount(self) -> None:
    """Populate columns when the application starts."""
    force = self._background_tasks and self.board.force_refresh_on_startup
    self.scan(initial=True, force=force)

def refresh_board(self, columns: Iterable[Column] | None = None) -> None:
    """Refresh columns without forcing their items to update external state."""
    self.scan(columns)

def scan(
    self,
    columns: Iterable[Column] | None = None,
    *,
    initial: bool = False,
    force: bool = False,
) -> None:
    """Recompute columns data in the background.

    A single scan feeds all columns: each item is read once,
    by a small pool of threads, and the resulting rows are dispatched
    to every column as they arrive. Each completed scan saves the displayed
    board to the cache when background tasks are enabled.

    Parameters:
        columns: The columns to update (all of them by default).
        initial: Whether this is the initial scan, which can display cached data.
        force: Whether to use the board's forced item hook.
    """
    if self._scanning:
        return
    self._scanning = True
    column_list = list(columns) if columns is not None else list(self.query(Column))
    self.run_worker(partial(self._scan, column_list, initial=initial, force=force), thread=True)

def compose(self) -> ComposeResult:
    """Screen composition."""
    yield VerticalScroll(Static(self.text), id="modal-contents")

def on_key(self, event: Key) -> None:
    """Dismiss on any unbound key."""
    if event.key not in self.app.active_bindings:
        event.stop()
        self.dismiss()

def modal(self, text: RenderableType) -> None:
    """Ask the UI thread to show text or a Rich renderable in a modal."""
    self.app.call_later(self._push_modal, text)