Helpers

helpers

Compatibility exports for the download helper layer.

DownloadReporter

Bases: ABC

Abstract progress reporter for file downloads.

add_file abstractmethod

add_file(name: str, total: Optional[int]) -> Any

Register a file and return an opaque handle used for later updates.

Source code in eos_downloader/helpers/progress.py

@abstractmethod
def add_file(self, name: str, total: Optional[int]) -> Any:
    """Register a file and return an opaque handle used for later updates."""

advance abstractmethod

advance(handle: Any, n_bytes: int) -> None

Report that n_bytes more bytes of handle have been written.

Source code in eos_downloader/helpers/progress.py

@abstractmethod
def advance(self, handle: Any, n_bytes: int) -> None:
    """Report that ``n_bytes`` more bytes of ``handle`` have been written."""

complete abstractmethod

complete(handle: Any) -> None

Mark the file identified by handle as finished.

Source code in eos_downloader/helpers/progress.py

@abstractmethod
def complete(self, handle: Any) -> None:
    """Mark the file identified by ``handle`` as finished."""

NoopReporter

Bases: DownloadReporter

Reporter that renders nothing.

PlainReporter

PlainReporter()

Bases: DownloadReporter

ANSI-free reporter emitting loguru lifecycle lines.

Source code in eos_downloader/helpers/progress.py

def __init__(self) -> None:
    self._files: Dict[int, Dict[str, Any]] = {}
    self._counter = 0
    self._lock = Lock()

RichReporter

RichReporter(console: Console, title: str = 'Downloading')

Bases: DownloadReporter

Animated grouped display for interactive terminals.

Source code in eos_downloader/helpers/progress.py

def __init__(self, console: Console, title: str = "Downloading") -> None:
    self._progress = Progress(
        SpinnerColumn(),
        TextColumn("[bold blue]{task.description}"),
        BarColumn(bar_width=None),
        TaskProgressColumn(),
        "•",
        DownloadColumn(),
        "•",
        TransferSpeedColumn(),
        "•",
        TimeRemainingColumn(),
        console=console,
    )
    self._live = Live(
        Panel(self._progress, title=title, border_style="blue"),
        console=console,
        refresh_per_second=10,
    )
    self._total_task = self._progress.add_task("Total", total=0)
    self._total_known = 0
    self._total_indeterminate = False
    self._lock = Lock()

download_files_concurrently

download_files_concurrently(
    items: Iterable[DownloadItem],
    reporter: DownloadReporter,
    *,
    max_workers: int = 4,
    block_size: int = 1024
) -> None

Download several files in parallel, updating a single shared reporter.

Source code in eos_downloader/helpers/transfer.py

def download_files_concurrently(
    items: Iterable[DownloadItem],
    reporter: DownloadReporter,
    *,
    max_workers: int = 4,
    block_size: int = 1024,
) -> None:
    """Download several files in parallel, updating a single shared reporter."""
    file_list: List[DownloadItem] = list(items)
    if not file_list:
        return
    done_event = Event()
    with sigint_guard(done_event), reporter:
        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            futures = [
                pool.submit(_stream_to_file, item, reporter, done_event, block_size)
                for item in file_list
            ]
            for future in futures:
                future.result()

resolve_reporter

resolve_reporter(
    mode: ProgressMode,
    console: Console,
    *,
    title: str = "Downloading"
) -> DownloadReporter

Build the concrete reporter for a progress mode and console.

Source code in eos_downloader/helpers/progress.py

def resolve_reporter(
    mode: ProgressMode,
    console: Console,
    *,
    title: str = "Downloading",
) -> DownloadReporter:
    """Build the concrete reporter for a progress ``mode`` and ``console``."""
    if mode == "none":
        return NoopReporter()
    if mode == "rich":
        use_rich = True
    elif mode == "plain":
        use_rich = False
    else:
        use_rich = console.is_terminal
    if use_rich:
        return RichReporter(console, title=title)
    return PlainReporter()

sigint_guard

sigint_guard(done_event: Event) -> Iterator[None]

Route SIGINT to a done event, then defer to the previous handler.

Source code in eos_downloader/helpers/signals.py

@contextmanager
def sigint_guard(done_event: Event) -> Iterator[None]:
    """Route ``SIGINT`` to a done event, then defer to the previous handler."""
    previous: Any = None
    installed = False

    def handler(signum: int, frame: Any) -> None:
        done_event.set()
        if callable(previous):
            previous(signum, frame)
        elif previous == signal.SIG_DFL:
            signal.default_int_handler(signum, frame)

    try:
        previous = signal.getsignal(signal.SIGINT)
        signal.signal(signal.SIGINT, handler)
        installed = True
    except ValueError:
        installed = False
    try:
        yield
    finally:
        if installed:
            signal.signal(signal.SIGINT, previous)