CLI

Utilities for command-line interfaces, including progress reporting for long-running pipeline stages.

ProgressTracker

ProgressTracker provides a TTY-aware progress bar designed for use inside ETL pipelines and ITask implementations. It detects sys.stdout.isatty() once at construction and routes output accordingly, animating on real terminals and falling back to structured logger.info calls on Lambda, ECS, or CI environments where \r overwriting is not meaningful.

Why not tqdm?

tqdm is the de-facto standard for progress bars in Python, but it introduces an external dependency that is unnecessary for the narrow use case this library covers. core-mixins is a foundational utility library, one of its explicit goals is to remain dependency-free (beyond the typing-extensions shim for Python < 3.11). Adding tqdm would propagate it as a transitive dependency into every project that depends on core-mixins, including production services running on AWS Lambda or ECS where package size and supply-chain surface area both matter.

ProgressTracker deliberately covers only what ETL pipelines stages need:

• A single animated line that updates in place on a real terminal.

• A persistent completion line with a summary note.

• Structured logging as the fallback when there is no terminal.

It does not aim to replace tqdm in interactive notebooks, download bars, or multi-bar scenarios. For those use cases, tqdm remains the right choice.

Progress bar live animation (TTY)

Completion with all stages passed (TTY)

Completion with validation errors (TTY)

CloudWatch output (Lambda / ECS / non-TTY)

On non-TTY environments progress() is silent and done() emits one structured logger.info line per stage, no \r noise in CloudWatch.

Routing behaviour

Environment	progress()	done()
TTY (real terminal)	\r animation	✔ line to stdout
Non-TTY (Lambda / ECS / CI)	silent	logger.info if set

Usage with ITask

Each pipeline stage is a concrete ITask subclass. The ProgressTracker is instantiated inside execute() with self.logger, so output routing (TTY animation vs. CloudWatch logs) is handled automatically.

import random
import time

from core_mixins.cli.progress_tracker import BOLD, GREEN, RESET, YELLOW, ProgressTracker
from core_mixins.interfaces.task import ITask
from core_mixins.logger import get_logger


class IngestTask(ITask):

    def execute(self, *args, **kwargs) -> dict:
        files = 120
        total_mb = 0.0
        t0 = time.monotonic()
        tracker = ProgressTracker(logger=self.logger)

        for i in range(1, files + 1):
            time.sleep(random.uniform(0.01, 0.025))
            total_mb += random.uniform(0.4, 3.2)
            elapsed = time.monotonic() - t0
            pct = i / files * 100
            tracker.progress("Ingesting files", pct, f"{total_mb / elapsed:.1f} MB/s")

        elapsed = time.monotonic() - t0
        tracker.done("Ingesting files", f"{files} files · {total_mb:.0f} MB · {elapsed:.1f}s")
        return {"records": files}


class ValidateTask(ITask):

    def execute(self, *args, **kwargs) -> dict:
        records = kwargs["records"]
        errors = 0
        t0 = time.monotonic()
        tracker = ProgressTracker(logger=self.logger)

        for i in range(1, records + 1):
            time.sleep(random.uniform(0.003, 0.009))
            if random.random() < 0.015:
                errors += 1
            pct = i / records * 100
            tracker.progress("Validating records", pct, f"{errors} errors")

        elapsed = time.monotonic() - t0
        tracker.done("Validating records", f"{records} records · {errors} errors · {elapsed:.1f}s")
        return {"records": records, "errors": errors}


if __name__ == "__main__":
    logger = get_logger(
        __name__,
        formatter="%(message)s",
        reset_handlers=True,
        propagate=False,
    )

    logger.info(f"\n{BOLD}  Pipeline starting…{RESET}\n")
    t0 = time.monotonic()

    ingest   = IngestTask(logger=logger).execute()
    validate = ValidateTask(logger=logger).execute(records=ingest["records"])

    elapsed = time.monotonic() - t0
    errors = validate["errors"]
    verdict = (
        f"{GREEN}{BOLD}  All stages completed in {elapsed:.1f}s.{RESET}"
        if errors == 0
        else f"{YELLOW}{BOLD}  Done in {elapsed:.1f}s — {errors} validation error(s).{RESET}"
    )
    logger.info(f"\n{verdict}\n")

API Reference

A lightweight terminal progress bar utility for CLI pipelines, with no external dependencies, pure stdlib.

class core_mixins.cli.progress_tracker.ProgressTracker(logger: Logger | None = None)

Bases: object

Progress tracker utility for CLI pipelines.

__init__(logger: Logger | None = None) → None

progress(desc: str, pct: float, note: str = '') → None

Overwrite the current terminal line.

done(desc: str, note: str = '') → None

Clear the progress line and print a persistent ✔ line.