Tracking Git progress with ProgressBar

[noonedeadpunk]

Hey folks!

I am really sorry for the stupid question, as I guess it's obvious for most of power users, but I'm trying to start a hobby project with textual and got stuck with trying to render git clone progress.

Like I managed quite fast to implement that usecase with tqdm:

import git
from git import RemoteProgress
from tqdm import tqdm

REPO_URL = "https://github.com/Textualize/textual"
REPO_PATH = "/tmp/textual"
BRANCH = "main"

class CloneProgress(RemoteProgress):
    def __init__(self):
        super().__init__()
        self.pbar = tqdm()

    def update(self, op_code, cur_count, max_count=None, message=''):
        self.pbar.total = max_count
        self.pbar.n = cur_count
        self.pbar.refresh()

if __name__ == "__main__":
    git.Repo.clone_from(REPO_URL, REPO_PATH, branch=BRANCH, progress=CloneProgress())

But I'm struggling with ProgressBar implementation so far. The best result I could reach to get bar to the resulting state after clone is completed:

from textual.app import App, ComposeResult
from textual.containers import Center, Middle
from textual.widgets import Footer, ProgressBar

import git
from git import RemoteProgress

REPO_URL = "https://github.com/Textualize/textual"
REPO_PATH = "/tmp/textual"
BRANCH = "main"

class IndeterminateProgressBar(App):
    BINDINGS = [("s", "start", "Start")]

    def compose(self) -> ComposeResult:
        with Center():
            with Middle():
                yield ProgressBar()
        yield Footer()

    def action_start(self) -> None:
        """Start the progress tracking."""
        widget = self.query_one(ProgressBar)
        ## I tried to add some initial update to the widget, but this also does not work
        # widget.update(total=100, progress=1)
        git.Repo.clone_from(REPO_URL, REPO_PATH, branch=BRANCH, progress=CloneProgress(widget))

class CloneProgress(RemoteProgress):
    def __init__(self, widget):
        super().__init__()
        self.pbar = widget

    def update(self, op_code, cur_count, max_count=None, message=''):
        self.pbar.update(total=max_count, progress=cur_count)

if __name__ == "__main__":
    IndeterminateProgressBar().run()

I do understand that it's due to git.Repo.clone_from is blocking process. But also, I am very confused on why widget.update(total=100, progress=1) does not render right away before git.Repo.clone_from - as I'd expect progress bar being updated right away, as these attributes seem to be reactive.

And I actually have quite some issues with updating text/data on the fly even when it's reactive, as it's always wait for the page to complete rendering first, so hoping that understanding this issue will help me a lot in other places as well (for instance I had a case where I needed to load 2 URIs, and wanted to print out current state on the screen so user won't be too confused waiting).

So it seems I'm missing a very crucial part on how to get screen update while waiting for some command/action.

[noonedeadpunk]

Ok, after switching over multiple AI models, I finally figured out how to do that. The working example is below:

import asyncio
import git
from git import RemoteProgress

from textual.app import App, ComposeResult
from textual.widgets import ProgressBar, Static, Footer
from textual.containers import Center, Middle
from textual.reactive import reactive

import os
import shutil

REPO_URL = "https://github.com/Textualize/textual"
REPO_PATH = "/tmp/textual"
BRANCH = "main"


class GitCloneProgress(RemoteProgress):
    def __init__(self, update_callback):
        super().__init__()
        self.update_callback = update_callback

    def update(self, op_code, cur_count, max_count=None, message=''):
        op_id = op_code & self.OP_MASK
        stage = ""
        if op_id == self.COUNTING:
            stage = "Counting objects"
        elif op_id == self.COMPRESSING:
            stage = "Compressing objects"
        elif op_id == self.RECEIVING:
            stage = "Receiving objects"
        elif op_id == self.RESOLVING:
            stage = "Resolving deltas"
        elif op_id == self.FINDING_SOURCES:
            stage = "Finding sources"
        elif op_id == self.REPORTING:
            stage = "Reporting"
        else:
            # Fallback for unknown operation IDs.
            stage = f"Processing ({op_id})"

        if max_count:
            self.update_callback(cur_count, max_count, stage)


class CloneApp(App):
    BINDINGS = [("s", "start", "Start")]
    progress = reactive(0)
    max_progress = reactive(100)

    def compose(self) -> ComposeResult:
        with Center():
            with Middle():
                yield Static("Cloning Repository...")
                yield ProgressBar(total=self.max_progress, show_percentage=True, id="progress-bar", show_eta=False)
        yield Footer()

    async def action_start(self) -> None:
        progress_bar = self.query_one(ProgressBar)
        static_message = self.query_one(Static)

        if os.path.exists(REPO_PATH):
            shutil.rmtree(REPO_PATH)

        def update_bar(cur, max_, stage):
            self.progress = cur
            self.max_progress = max_
            progress_bar.update(total=max_, progress=cur)
            static_message.update(stage)

        loop = asyncio.get_event_loop()
        await loop.run_in_executor(
            None,
            lambda: git.Repo.clone_from(
                REPO_URL,
                REPO_PATH,
                branch=BRANCH,
                progress=GitCloneProgress(update_bar)
            )
        )

        progress_bar.update(total=self.max_progress, progress=self.max_progress)
        static_message.update("Clone Complete")


if __name__ == "__main__":
    app = CloneApp()
    app.run()