devboard ¤

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_apply(self, action: str = "default") -> None:
    """Apply an action to selected rows."""
    selected_rows = list(self.table.selected_rows) or [self.table.current_row]
    if self.THREADED:
        for row in selected_rows:
            self.run_worker(partial(self.apply, action=action, row=row), thread=True)  # type: ignore[arg-type]
    else:
        for row in selected_rows:
            self.apply(action=action, row=row)  # type: ignore[arg-type]

def apply(self, action: str, row: Row) -> None:  # noqa: ARG002
    """Apply action on given row."""
    return

def compose(self) -> ComposeResult:
    """Compose column widgets."""
    yield Static("▶ " + self.TITLE, classes="column-title")
    yield DataTable(id="table")

def list_projects(self) -> Iterable[Project]:
    """List projects for this column."""
    return ()

def modal(self, text: str) -> None:
    """Push a modal."""
    self.app.push_screen(Modal(text=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 on_mount(self) -> None:
    """Fill data table."""
    self.update()

@staticmethod
def populate_rows(project: Project) -> list[tuple[Any, ...]]:  # noqa: ARG004
    """Populate rows for this column."""
    return []

def update(self) -> None:
    """Update the column (recompute data)."""
    table = self.query_one(DataTable)
    if table.loading:
        return
    table.loading = True
    table.clear(columns=True)
    table.cursor_type = "row"
    self._load_data(table)

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_rows(self, rows: Iterable[Iterable]) -> list[RowKey]:
    """Add rows.

    Automatically insert a column with checkboxes in position 0.
    """
    return super().add_rows((Checkbox(), *row) for row in rows)

def clear(self, columns: bool = True) -> SelectableRowsDataTable:  # noqa: FBT001,FBT002
    """Clear rows and optionally columns.

    When clearing columns, automatically re-add a column for checkboxes.
    """
    super().clear(columns)
    if columns:
        self.add_column("", key="checkbox")
    return self

def force_refresh(self) -> None:
    """Force refresh table."""
    # HACK: Without such increment, the table is refreshed
    # only when focus changes to another column.
    self._update_count += 1
    self.refresh()

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

def action_refresh(self) -> None:
    """Refresh all columns."""
    for column in self.query(Column):
        column.update()

def action_show_help(self) -> None:
    """Show help."""
    lines = ["# Main keys\n\n"]
    lines.extend(self._bindings_help(Devboard))
    lines.extend(self._bindings_help(DataTable, search_up=True))
    for column in self.query(Column):
        lines.append(f"\n\n# {column.__class__.TITLE}\n\n")
        lines.extend(self._bindings_help(column.__class__))
    self.push_screen(Modal(text=Markdown("\n".join(lines))))

def compose(self) -> ComposeResult:
    """Compose the layout."""
    for column in self._load_columns():
        if isinstance(column, Column):
            yield column
        else:
            yield column()
    yield Footer()

@work(thread=True)
def fetch_all(self) -> None:
    """Run `git fetch` in all projects, in background."""
    projects = set()
    for column in self.query(Column):
        projects |= set(column.list_projects())
    with Pool() as pool:
        for project in projects:
            pool.apply_async(project.fetch)

def modal(self, text: str) -> None:
    """Push a modal."""
    self.app.push_screen(Modal(text=text))

def on_mount(self) -> None:
    """Run background tasks."""
    if self._background_tasks:
        self.fetch_all()

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 not any(bindings.keys.get(event.key) for _, bindings in self.app._modal_binding_chain):  # type: ignore[attr-defined]
        self.dismiss()

def modal(self, text: str) -> None:
    """Push a modal."""
    self.app.push_screen(Modal(text=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)

@contextmanager
def checkout(self, branch: str | None) -> Iterator[None]:
    """Checkout branch, restore previous one when exiting."""
    if not branch:
        yield
        return
    current = self.branch
    if branch == current:
        yield
        return
    self.repo.branches[branch].checkout()
    try:
        yield
    finally:
        current.checkout()

def delete(self, branch: str) -> None:
    """Delete branch."""
    self.repo.delete_head(branch, force=True)

def fetch(self) -> None:
    """Fetch."""
    with suppress(AttributeError, GitCommandError):
        self.repo.remotes.origin.fetch()
    with suppress(AttributeError, GitCommandError):
        self.repo.remotes.upstream.fetch()

def lock(self) -> bool:
    """Lock project."""
    return self.LOCKS[self].acquire(blocking=False)

def pull(self, branch: str | None = None) -> None:
    """Pull branch."""
    with self.checkout(branch):
        self.repo.remotes.origin.pull()

def push(self, branch: str | None = None) -> None:
    """Push branch."""
    with self.checkout(branch):
        self.repo.remotes.origin.push()

def unlock(self) -> None:
    """Unlock project."""
    self.LOCKS[self].release()

def unpulled(self, remote: str = "origin") -> dict[str, int]:
    """Number of unpulled commits, per branch."""
    result = {}
    for branch in self.repo.branches:
        with contextlib.suppress(GitCommandError):
            result[branch.name] = len(list(self.repo.iter_commits(f"{branch.name}..{remote}/{branch.name}")))
    return result

def unpushed(self, remote: str = "origin") -> dict[str, int]:
    """Number of unpushed commits, per branch."""
    result = {}
    for branch in self.repo.branches:
        with contextlib.suppress(GitCommandError):
            result[branch.name] = len(list(self.repo.iter_commits(f"{remote}/{branch.name}..{branch.name}")))
    return result

def unreleased(self, branch: str | None = None) -> list[Commit]:
    """List unreleased commits."""
    commits = []
    if branch is None:
        try:
            branch = self.default_branch
        except ValueError:
            return []
    iterator = self.repo.iter_commits(branch)
    try:
        latest_tagged_commit = self.latest_tag.commit
    except IndexError:
        return list(iterator)
    for commit in iterator:
        if commit == latest_tagged_commit:
            break
        commits.append(commit)
    return commits

def remove(self) -> None:
    """Remove row from the table."""
    self.table.remove_row(self.key)

def select(self) -> None:
    """Select this row."""
    self.checkbox.check()

def toggle_select(self) -> bool:
    """Toggle-select this row."""
    return self.checkbox.toggle()

def unselect(self) -> None:
    """Unselect this row."""
    self.checkbox.uncheck()

def remove(self) -> None:
    """Remove row from the table."""
    self.table.remove_row(self.key)

def select(self) -> None:
    """Select this row."""
    self.checkbox.check()

def toggle_select(self) -> bool:
    """Toggle-select this row."""
    return self.checkbox.toggle()

def unselect(self) -> None:
    """Unselect this row."""
    self.checkbox.uncheck()

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_rows(self, rows: Iterable[Iterable]) -> list[RowKey]:
    """Add rows.

    Automatically insert a column with checkboxes in position 0.
    """
    return super().add_rows((Checkbox(), *row) for row in rows)

def clear(self, columns: bool = True) -> SelectableRowsDataTable:  # noqa: FBT001,FBT002
    """Clear rows and optionally columns.

    When clearing columns, automatically re-add a column for checkboxes.
    """
    super().clear(columns)
    if columns:
        self.add_column("", key="checkbox")
    return self

def force_refresh(self) -> None:
    """Force refresh table."""
    # HACK: Without such increment, the table is refreshed
    # only when focus changes to another column.
    self._update_count += 1
    self.refresh()