Progress API Reference

Complete API reference for Concurry's progress tracking.

Module: concurry.utils.progress

Classes

ProgressBar

A progress bar implementation using tqdm with additional features.

This class provides beautiful, informative progress bars with rich features including: - Automatic success/failure/stop state indicators with color coding - Multiple styles (auto, notebook, standard, Ray) - Iterable wrapping for easy integration - Fine-grained control over updates - Customizable appearance

Example

from concurry.utils.progress import ProgressBar
import time

# Wrap an iterable
for item in ProgressBar(range(100), desc="Processing"):
    time.sleep(0.01)
# Automatically shows success!

# Manual progress bar
pbar = ProgressBar(total=100, desc="Manual")
for i in range(100):
    time.sleep(0.01)
    pbar.update(1)
pbar.success("Complete!")

Source code in src/concurry/utils/progress.py

class ProgressBar:
    """A progress bar implementation using tqdm with additional features.

    This class provides beautiful, informative progress bars with rich features including:
    - Automatic success/failure/stop state indicators with color coding
    - Multiple styles (auto, notebook, standard, Ray)
    - Iterable wrapping for easy integration
    - Fine-grained control over updates
    - Customizable appearance

    Example:
        ```python
        from concurry.utils.progress import ProgressBar
        import time

        # Wrap an iterable
        for item in ProgressBar(range(100), desc="Processing"):
            time.sleep(0.01)
        # Automatically shows success!

        # Manual progress bar
        pbar = ProgressBar(total=100, desc="Manual")
        for i in range(100):
            time.sleep(0.01)
            pbar.update(1)
        pbar.success("Complete!")
        ```
    """

    def __init__(
        self,
        *args,
        pbar: Optional[TqdmProgressBar] = None,
        style: Literal["auto", "notebook", "std", "ray"] = "auto",
        unit: str = "row",
        color: str = "#0288d1",  # Bluish
        ncols: Optional[int] = None,
        smoothing: Optional[float] = None,
        total: Optional[int] = None,
        disable: bool = False,
        miniters: Optional[int] = None,
        progress_bar: Union[bool, dict, "ProgressBar"] = True,
        prefer_kwargs: bool = True,
        **kwargs,
    ):
        """Initialize a progress bar.

        Args:
            pbar: Optional existing tqdm progress bar instance
            style: Progress bar style. Options:
                - "auto": Automatically detect environment (default)
                - "notebook": Optimized for Jupyter notebooks
                - "std": Standard terminal output
                - "ray": Ray distributed progress (requires Ray)
            unit: Unit name for items being processed (default: "row")
            color: Hex color code for the progress bar (default: "#0288d1" - blue)
            ncols: Width of the progress bar in characters (default: global_config.defaults.progress_bar_ncols)
            smoothing: Smoothing factor for progress updates (default: global_config.defaults.progress_bar_smoothing)
            total: Total number of items to process
            disable: Whether to disable the progress bar (default: False)
            miniters: Minimum iterations between display updates (default: global_config.defaults.progress_bar_miniters)
                Higher values improve performance for large iteration counts
            progress_bar: Progress bar configuration:
                - True: Use default configuration
                - False/None: Disable progress bar
                - dict: Configuration dictionary
                - ProgressBar: Existing ProgressBar instance to reuse
            prefer_kwargs: Whether to prefer kwargs over progress_bar dict (default: True)
            **kwargs: Additional arguments to pass to tqdm

        Example:
            ```python
            # Basic usage
            pbar = ProgressBar(total=100, desc="Processing")

            # Custom colors and styling
            pbar = ProgressBar(
                total=100,
                desc="Custom",
                unit="files",
                color="#9c27b0",  # Purple
                ncols=120,
                miniters=5
            )

            # Disable for non-interactive environments
            pbar = ProgressBar(total=100, disable=True)
            ```
        """
        # Initialize _extra_fields first using object.__setattr__
        object.__setattr__(self, "_extra_fields", {})

        # Load defaults from global config if not provided
        from ..config import global_config

        local_config = global_config.clone()
        if ncols is None:
            ncols = local_config.defaults.progress_bar_ncols
        if smoothing is None:
            smoothing = local_config.defaults.progress_bar_smoothing
        if miniters is None:
            miniters = local_config.defaults.progress_bar_miniters

        # Handle progress_bar parameter
        if isinstance(progress_bar, ProgressBar):
            if prefer_kwargs:
                if "total" in kwargs:
                    progress_bar.set_total(kwargs["total"])
                if "initial" in kwargs:
                    progress_bar.set_n(kwargs["initial"])
                if "desc" in kwargs:
                    progress_bar.set_description(kwargs["desc"])
                if "unit" in kwargs:
                    progress_bar.set_unit(kwargs["unit"])
            self.__dict__.update(progress_bar.__dict__)
            return

        if progress_bar is not None and not isinstance(progress_bar, (bool, dict)):
            raise ValueError(
                f"You must pass `progress_bar` as either a bool, dict or None (None or False disables it). "
                f"Found: {type(progress_bar)}"
            )

        if progress_bar is True:
            progress_bar = dict()
        elif progress_bar is False:
            progress_bar = None

        if progress_bar is not None and not isinstance(progress_bar, dict):
            raise ValueError(
                "You must pass `progress_bar` as either a bool, dict or None. None or False disables it."
            )

        if progress_bar is None:
            progress_bar = dict(disable=True)
        elif isinstance(progress_bar, dict) and len(kwargs) > 0:
            if prefer_kwargs is True:
                progress_bar = {
                    **progress_bar,
                    **kwargs,
                }
            else:
                progress_bar = {
                    **kwargs,
                    **progress_bar,
                }

        # Set instance attributes using object.__setattr__ to avoid recursion
        object.__setattr__(self, "pbar", pbar)
        object.__setattr__(self, "style", style)
        object.__setattr__(self, "unit", unit)
        object.__setattr__(self, "color", color)
        object.__setattr__(self, "ncols", ncols)
        object.__setattr__(self, "smoothing", smoothing)
        object.__setattr__(self, "total", total)
        object.__setattr__(self, "disable", disable)
        object.__setattr__(self, "miniters", miniters)
        object.__setattr__(self, "_pending_updates", 0)

        # Validate miniters
        if self.miniters < 1:
            raise ValueError("miniters must be greater than or equal to 1")

        # Store extra fields
        for field_name, field_value in progress_bar.items():
            if not hasattr(self, field_name):
                self._extra_fields[field_name] = field_value

        # Create progress bar with current settings
        pbar = self._create_pbar(
            **{
                k: v
                for k, v in {**self.__dict__, **self._extra_fields}.items()
                if k not in {"pbar", "color", "_pending_updates", "_extra_fields"}
            }
        )
        pbar.color = self.color
        pbar.refresh()
        self.pbar = pbar

    @classmethod
    def _create_pbar(
        cls,
        style: Literal["auto", "notebook", "std", "ray"],
        **kwargs,
    ) -> TqdmProgressBar:
        """Create a tqdm progress bar with the specified style."""
        if style == "auto":
            if _IS_IPYWIDGETS_INSTALLED:
                kwargs["ncols"]: Optional[int] = None
            return AutoTqdmProgressBar(**kwargs)
        elif style == "notebook":
            if _IS_IPYWIDGETS_INSTALLED:
                kwargs["ncols"]: Optional[int] = None
            return NotebookTqdmProgressBar(**kwargs)
        elif _IS_RAY_INSTALLED and style == "ray":
            from ray.experimental import tqdm_ray

            kwargs = {k: v for k, v in kwargs.items() if k in get_args_and_kwargs(tqdm_ray.tqdm)}
            return tqdm_ray.tqdm(**kwargs)
        else:
            return StdTqdmProgressBar(**kwargs)

    def __new__(cls, *args, **kwargs):
        """Handle both direct instantiation and iterable wrapping."""
        # If first argument is an iterable (but not a string), use iter
        if (
            len(args) > 0
            and hasattr(args[0], "__iter__")
            and not isinstance(args[0], (str, bytes, bytearray))
        ):
            return cls._iter(args[0], **kwargs)
        return super().__new__(cls)

    @classmethod
    def _iter(
        cls,
        iterable: Union[Generator, Iterator, List, Tuple, Set, Dict, ItemsView],
        **kwargs,
    ):
        """Create a progress bar that wraps an iterable."""
        if isinstance(iterable, (list, tuple, dict, ItemsView, set, frozenset)):
            kwargs["total"] = len(iterable)
        if isinstance(iterable, dict):
            iterable: ItemsView = iterable.items()

        pbar = cls(**kwargs)

        try:
            for item in iterable:
                yield item
                pbar.update(1)
            pbar.success()
        except Exception as e:
            pbar.failure()
            raise e

    def update(self, n: int = 1) -> Optional[bool]:
        """Update the progress bar by n steps.

        Updates are batched based on the `miniters` parameter for better performance.
        The display only updates when enough iterations have accumulated.

        Args:
            n: Number of steps to increment (default: 1)

        Returns:
            True if the display was updated, None if the update was buffered

        Example:
            ```python
            pbar = ProgressBar(total=100)
            for i in range(100):
                # Do work
                pbar.update(1)
            ```
        """
        self._pending_updates += n
        if abs(self._pending_updates) >= self.miniters:
            out = self.pbar.update(n=self._pending_updates)
            self.refresh()
            self._pending_updates = 0
            return out
        else:
            return None

    def set_n(self, new_n: int) -> None:
        """Set the current progress value directly.

        This method sets the progress to an absolute value rather than incrementing.
        Useful for jumping to a specific point in the progress.

        Args:
            new_n: The new progress value

        Example:
            ```python
            pbar = ProgressBar(total=100)
            pbar.set_n(50)  # Jump to 50% complete
            ```
        """
        self.pbar.update(n=new_n - self.pbar.n)
        self._pending_updates = 0  # Clear all updates after setting new value
        self.refresh()

    def set_total(self, new_total: int) -> None:
        """Set the total number of items to process.

        This method allows dynamically updating the total when the final count
        becomes known or changes during processing.

        Args:
            new_total: The new total value

        Example:
            ```python
            pbar = ProgressBar(total=100)
            # ... process 50 items ...
            # More work discovered!
            pbar.set_total(150)
            ```
        """
        self.pbar.total = new_total
        self._pending_updates = 0  # Clear all updates after setting new value
        self.refresh()

    def set_description(self, desc: Optional[str] = None, refresh: Optional[bool] = True) -> Optional[str]:
        """Set the description of the progress bar.

        This method allows dynamically updating the description text during processing,
        useful for showing different phases or stages.

        Args:
            desc: New description text
            refresh: Whether to refresh the display (default: True)

        Returns:
            The previous description

        Example:
            ```python
            pbar = ProgressBar(total=100, desc="Phase 1")
            for i in range(50):
                pbar.update(1)
            pbar.set_description("Phase 2")
            for i in range(50):
                pbar.update(1)
            ```
        """
        out = self.pbar.set_description(desc=desc, refresh=refresh)
        self.refresh()
        return out

    def set_unit(self, new_unit: str) -> None:
        """Set the unit displayed in the progress bar.

        This method allows changing the unit name during processing, useful when
        switching between different types of work (e.g., files to MB).

        Args:
            new_unit: The new unit name (e.g., "files", "MB", "items")

        Example:
            ```python
            pbar = ProgressBar(total=100, unit="files")
            # ... process files ...
            pbar.set_unit("MB")  # Switch to showing MB processed
            ```
        """
        self.pbar.unit = new_unit
        self.refresh()

    def success(self, desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None:
        """Mark the progress bar as successful (green color).

        This method completes the progress bar with a success indicator, changing
        the color to green and optionally adding a success message.

        Args:
            desc: Success message to display
            close: Whether to close the progress bar (default: True)
            append_desc: Whether to append to existing description (default: True)

        Example:
            ```python
            pbar = ProgressBar(total=100, desc="Processing")
            for i in range(100):
                pbar.update(1)
            pbar.success("All done!")  # Shows green bar with message
            ```
        """
        self._complete_with_status(
            color="#43a047",  # Dark Green
            desc=desc,
            close=close,
            append_desc=append_desc,
        )

    def stop(self, desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None:
        """Mark the progress bar as stopped (grey color).

        This method completes the progress bar with a stop indicator, changing
        the color to grey and optionally adding a stop message. Useful for
        indicating early termination or cancellation.

        Args:
            desc: Stop message to display
            close: Whether to close the progress bar (default: True)
            append_desc: Whether to append to existing description (default: True)

        Example:
            ```python
            pbar = ProgressBar(total=100, desc="Processing")
            for i in range(100):
                if should_stop():
                    pbar.stop("Cancelled by user")  # Shows grey bar
                    break
                pbar.update(1)
            ```
        """
        self._complete_with_status(
            color="#b0bec5",  # Dark Grey
            desc=desc,
            close=close,
            append_desc=append_desc,
        )

    def failure(self, desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None:
        """Mark the progress bar as failed (red color).

        This method completes the progress bar with a failure indicator, changing
        the color to red and optionally adding a failure message. Use in exception
        handlers to indicate errors.

        Args:
            desc: Failure message to display
            close: Whether to close the progress bar (default: True)
            append_desc: Whether to append to existing description (default: True)

        Example:
            ```python
            pbar = ProgressBar(total=100, desc="Processing")
            try:
                for i in range(100):
                    if error_occurred():
                        raise ValueError("Error")
                    pbar.update(1)
                pbar.success()
            except Exception as e:
                pbar.failure(f"Failed: {e}")  # Shows red bar
                raise
            ```
        """
        self._complete_with_status(
            color="#e64a19",  # Dark Red
            desc=desc,
            close=close,
            append_desc=append_desc,
        )

    def _complete_with_status(
        self,
        color: str,
        desc: Optional[str],
        close: bool,
        append_desc: bool,
    ) -> None:
        """Complete the progress bar with a status."""
        if getattr(self.pbar, "disable", None) is False:
            self.pbar.update(n=self._pending_updates)
            self._pending_updates = 0
            self.color = color
            self.pbar.colour = color
            if desc is not None:
                if append_desc:
                    desc: str = f"[{desc}] {self.pbar.desc}"
                self.pbar.desc = desc
            self.pbar.refresh()
            if close:
                self.close()

    def refresh(self) -> None:
        """Refresh the progress bar display.

        This method manually updates the progress bar display. It's typically
        called automatically by other methods, but can be called manually if needed.

        Example:
            ```python
            pbar = ProgressBar(total=100)
            pbar.color = "#9c27b0"  # Change color
            pbar.refresh()  # Apply the change
            ```
        """
        self.pbar.colour = self.color
        self.pbar.refresh()

    def close(self) -> None:
        """Close and clean up the progress bar.

        This method closes the progress bar and releases any resources. It's
        automatically called by success(), failure(), and stop() when close=True.

        Example:
            ```python
            pbar = ProgressBar(total=100)
            try:
                for i in range(100):
                    pbar.update(1)
            finally:
                pbar.close()  # Ensure cleanup
            ```
        """
        self.pbar.refresh()
        self.pbar.close()
        self.pbar.refresh()

    def __del__(self) -> None:
        """Clean up the progress bar when the object is deleted."""
        if hasattr(self, "pbar") and self.pbar is not None:
            self.pbar.close()

    def __getattr__(self, name: str) -> Any:
        """Handle access to extra fields."""
        # Use object.__getattribute__ to avoid recursion
        extra_fields = object.__getattribute__(self, "_extra_fields")
        if name in extra_fields:
            return extra_fields[name]
        raise AttributeError(f"'{self.__class__.__name__}' has no attribute '{name}'")

    def __setattr__(self, name: str, value: Any) -> None:
        """Handle setting extra fields."""
        # Use object.__getattribute__ to avoid recursion
        if name == "_extra_fields":
            object.__setattr__(self, name, value)
            return

        if name in self.__dict__ or name in self.__class__.__dict__:
            object.__setattr__(self, name, value)
        else:
            extra_fields = object.__getattribute__(self, "_extra_fields")
            extra_fields[name] = value

__init__(*args, pbar: Optional[TqdmProgressBar] = None, style: Literal['auto', 'notebook', 'std', 'ray'] = 'auto', unit: str = 'row', color: str = '#0288d1', ncols: Optional[int] = None, smoothing: Optional[float] = None, total: Optional[int] = None, disable: bool = False, miniters: Optional[int] = None, progress_bar: Union[bool, dict, ProgressBar] = True, prefer_kwargs: bool = True, **kwargs)

Parameters:

Name	Type	Description	Default
pbar	Optional[TqdmProgressBar]	Optional existing tqdm progress bar instance	None
style	Literal['auto', 'notebook', 'std', 'ray']	Progress bar style. Options: - "auto": Automatically detect environment (default) - "notebook": Optimized for Jupyter notebooks - "std": Standard terminal output - "ray": Ray distributed progress (requires Ray)	'auto'
unit	str	Unit name for items being processed (default: "row")	'row'
color	str	Hex color code for the progress bar (default: "#0288d1" - blue)	'#0288d1'
ncols	Optional[int]	Width of the progress bar in characters (default: global_config.defaults.progress_bar_ncols)	None
smoothing	Optional[float]	Smoothing factor for progress updates (default: global_config.defaults.progress_bar_smoothing)	None
total	Optional[int]	Total number of items to process	None
disable	bool	Whether to disable the progress bar (default: False)	False
miniters	Optional[int]	Minimum iterations between display updates (default: global_config.defaults.progress_bar_miniters) Higher values improve performance for large iteration counts	None
progress_bar	Union[bool, dict, ProgressBar]	Progress bar configuration: - True: Use default configuration - False/None: Disable progress bar - dict: Configuration dictionary - ProgressBar: Existing ProgressBar instance to reuse	True
prefer_kwargs	bool	Whether to prefer kwargs over progress_bar dict (default: True)	True
**kwargs		Additional arguments to pass to tqdm	{}

Example

# Basic usage
pbar = ProgressBar(total=100, desc="Processing")

# Custom colors and styling
pbar = ProgressBar(
    total=100,
    desc="Custom",
    unit="files",
    color="#9c27b0",  # Purple
    ncols=120,
    miniters=5
)

# Disable for non-interactive environments
pbar = ProgressBar(total=100, disable=True)

Source code in src/concurry/utils/progress.py

def __init__(
    self,
    *args,
    pbar: Optional[TqdmProgressBar] = None,
    style: Literal["auto", "notebook", "std", "ray"] = "auto",
    unit: str = "row",
    color: str = "#0288d1",  # Bluish
    ncols: Optional[int] = None,
    smoothing: Optional[float] = None,
    total: Optional[int] = None,
    disable: bool = False,
    miniters: Optional[int] = None,
    progress_bar: Union[bool, dict, "ProgressBar"] = True,
    prefer_kwargs: bool = True,
    **kwargs,
):
    """Initialize a progress bar.

    Args:
        pbar: Optional existing tqdm progress bar instance
        style: Progress bar style. Options:
            - "auto": Automatically detect environment (default)
            - "notebook": Optimized for Jupyter notebooks
            - "std": Standard terminal output
            - "ray": Ray distributed progress (requires Ray)
        unit: Unit name for items being processed (default: "row")
        color: Hex color code for the progress bar (default: "#0288d1" - blue)
        ncols: Width of the progress bar in characters (default: global_config.defaults.progress_bar_ncols)
        smoothing: Smoothing factor for progress updates (default: global_config.defaults.progress_bar_smoothing)
        total: Total number of items to process
        disable: Whether to disable the progress bar (default: False)
        miniters: Minimum iterations between display updates (default: global_config.defaults.progress_bar_miniters)
            Higher values improve performance for large iteration counts
        progress_bar: Progress bar configuration:
            - True: Use default configuration
            - False/None: Disable progress bar
            - dict: Configuration dictionary
            - ProgressBar: Existing ProgressBar instance to reuse
        prefer_kwargs: Whether to prefer kwargs over progress_bar dict (default: True)
        **kwargs: Additional arguments to pass to tqdm

    Example:
        ```python
        # Basic usage
        pbar = ProgressBar(total=100, desc="Processing")

        # Custom colors and styling
        pbar = ProgressBar(
            total=100,
            desc="Custom",
            unit="files",
            color="#9c27b0",  # Purple
            ncols=120,
            miniters=5
        )

        # Disable for non-interactive environments
        pbar = ProgressBar(total=100, disable=True)
        ```
    """
    # Initialize _extra_fields first using object.__setattr__
    object.__setattr__(self, "_extra_fields", {})

    # Load defaults from global config if not provided
    from ..config import global_config

    local_config = global_config.clone()
    if ncols is None:
        ncols = local_config.defaults.progress_bar_ncols
    if smoothing is None:
        smoothing = local_config.defaults.progress_bar_smoothing
    if miniters is None:
        miniters = local_config.defaults.progress_bar_miniters

    # Handle progress_bar parameter
    if isinstance(progress_bar, ProgressBar):
        if prefer_kwargs:
            if "total" in kwargs:
                progress_bar.set_total(kwargs["total"])
            if "initial" in kwargs:
                progress_bar.set_n(kwargs["initial"])
            if "desc" in kwargs:
                progress_bar.set_description(kwargs["desc"])
            if "unit" in kwargs:
                progress_bar.set_unit(kwargs["unit"])
        self.__dict__.update(progress_bar.__dict__)
        return

    if progress_bar is not None and not isinstance(progress_bar, (bool, dict)):
        raise ValueError(
            f"You must pass `progress_bar` as either a bool, dict or None (None or False disables it). "
            f"Found: {type(progress_bar)}"
        )

    if progress_bar is True:
        progress_bar = dict()
    elif progress_bar is False:
        progress_bar = None

    if progress_bar is not None and not isinstance(progress_bar, dict):
        raise ValueError(
            "You must pass `progress_bar` as either a bool, dict or None. None or False disables it."
        )

    if progress_bar is None:
        progress_bar = dict(disable=True)
    elif isinstance(progress_bar, dict) and len(kwargs) > 0:
        if prefer_kwargs is True:
            progress_bar = {
                **progress_bar,
                **kwargs,
            }
        else:
            progress_bar = {
                **kwargs,
                **progress_bar,
            }

    # Set instance attributes using object.__setattr__ to avoid recursion
    object.__setattr__(self, "pbar", pbar)
    object.__setattr__(self, "style", style)
    object.__setattr__(self, "unit", unit)
    object.__setattr__(self, "color", color)
    object.__setattr__(self, "ncols", ncols)
    object.__setattr__(self, "smoothing", smoothing)
    object.__setattr__(self, "total", total)
    object.__setattr__(self, "disable", disable)
    object.__setattr__(self, "miniters", miniters)
    object.__setattr__(self, "_pending_updates", 0)

    # Validate miniters
    if self.miniters < 1:
        raise ValueError("miniters must be greater than or equal to 1")

    # Store extra fields
    for field_name, field_value in progress_bar.items():
        if not hasattr(self, field_name):
            self._extra_fields[field_name] = field_value

    # Create progress bar with current settings
    pbar = self._create_pbar(
        **{
            k: v
            for k, v in {**self.__dict__, **self._extra_fields}.items()
            if k not in {"pbar", "color", "_pending_updates", "_extra_fields"}
        }
    )
    pbar.color = self.color
    pbar.refresh()
    self.pbar = pbar

update(n: int = 1) -> Optional[bool]

Update the progress bar by n steps.

Updates are batched based on the miniters parameter for better performance. The display only updates when enough iterations have accumulated.

Parameters:

Name	Type	Description	Default
n	int	Number of steps to increment (default: 1)	1

Returns:

Type	Description
Optional[bool]	True if the display was updated, None if the update was buffered

Example

pbar = ProgressBar(total=100)
for i in range(100):
    # Do work
    pbar.update(1)

Source code in src/concurry/utils/progress.py

def update(self, n: int = 1) -> Optional[bool]:
    """Update the progress bar by n steps.

    Updates are batched based on the `miniters` parameter for better performance.
    The display only updates when enough iterations have accumulated.

    Args:
        n: Number of steps to increment (default: 1)

    Returns:
        True if the display was updated, None if the update was buffered

    Example:
        ```python
        pbar = ProgressBar(total=100)
        for i in range(100):
            # Do work
            pbar.update(1)
        ```
    """
    self._pending_updates += n
    if abs(self._pending_updates) >= self.miniters:
        out = self.pbar.update(n=self._pending_updates)
        self.refresh()
        self._pending_updates = 0
        return out
    else:
        return None

set_n(new_n: int) -> None

Set the current progress value directly.

This method sets the progress to an absolute value rather than incrementing. Useful for jumping to a specific point in the progress.

Parameters:

Name	Type	Description	Default
new_n	int	The new progress value	required

Example

pbar = ProgressBar(total=100)
pbar.set_n(50)  # Jump to 50% complete

Source code in src/concurry/utils/progress.py

def set_n(self, new_n: int) -> None:
    """Set the current progress value directly.

    This method sets the progress to an absolute value rather than incrementing.
    Useful for jumping to a specific point in the progress.

    Args:
        new_n: The new progress value

    Example:
        ```python
        pbar = ProgressBar(total=100)
        pbar.set_n(50)  # Jump to 50% complete
        ```
    """
    self.pbar.update(n=new_n - self.pbar.n)
    self._pending_updates = 0  # Clear all updates after setting new value
    self.refresh()

set_total(new_total: int) -> None

Set the total number of items to process.

This method allows dynamically updating the total when the final count becomes known or changes during processing.

Parameters:

Name	Type	Description	Default
new_total	int	The new total value	required

Example

pbar = ProgressBar(total=100)
# ... process 50 items ...
# More work discovered!
pbar.set_total(150)

Source code in src/concurry/utils/progress.py

def set_total(self, new_total: int) -> None:
    """Set the total number of items to process.

    This method allows dynamically updating the total when the final count
    becomes known or changes during processing.

    Args:
        new_total: The new total value

    Example:
        ```python
        pbar = ProgressBar(total=100)
        # ... process 50 items ...
        # More work discovered!
        pbar.set_total(150)
        ```
    """
    self.pbar.total = new_total
    self._pending_updates = 0  # Clear all updates after setting new value
    self.refresh()

set_description(desc: Optional[str] = None, refresh: Optional[bool] = True) -> Optional[str]

Set the description of the progress bar.

This method allows dynamically updating the description text during processing, useful for showing different phases or stages.

Parameters:

Name	Type	Description	Default
desc	Optional[str]	New description text	None
refresh	Optional[bool]	Whether to refresh the display (default: True)	True

Returns:

Type	Description
Optional[str]	The previous description

Example

pbar = ProgressBar(total=100, desc="Phase 1")
for i in range(50):
    pbar.update(1)
pbar.set_description("Phase 2")
for i in range(50):
    pbar.update(1)

Source code in src/concurry/utils/progress.py

def set_description(self, desc: Optional[str] = None, refresh: Optional[bool] = True) -> Optional[str]:
    """Set the description of the progress bar.

    This method allows dynamically updating the description text during processing,
    useful for showing different phases or stages.

    Args:
        desc: New description text
        refresh: Whether to refresh the display (default: True)

    Returns:
        The previous description

    Example:
        ```python
        pbar = ProgressBar(total=100, desc="Phase 1")
        for i in range(50):
            pbar.update(1)
        pbar.set_description("Phase 2")
        for i in range(50):
            pbar.update(1)
        ```
    """
    out = self.pbar.set_description(desc=desc, refresh=refresh)
    self.refresh()
    return out

set_unit(new_unit: str) -> None

Set the unit displayed in the progress bar.

This method allows changing the unit name during processing, useful when switching between different types of work (e.g., files to MB).

Parameters:

Name	Type	Description	Default
new_unit	str	The new unit name (e.g., "files", "MB", "items")	required

Example

pbar = ProgressBar(total=100, unit="files")
# ... process files ...
pbar.set_unit("MB")  # Switch to showing MB processed

Source code in src/concurry/utils/progress.py

def set_unit(self, new_unit: str) -> None:
    """Set the unit displayed in the progress bar.

    This method allows changing the unit name during processing, useful when
    switching between different types of work (e.g., files to MB).

    Args:
        new_unit: The new unit name (e.g., "files", "MB", "items")

    Example:
        ```python
        pbar = ProgressBar(total=100, unit="files")
        # ... process files ...
        pbar.set_unit("MB")  # Switch to showing MB processed
        ```
    """
    self.pbar.unit = new_unit
    self.refresh()

success(desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None

Mark the progress bar as successful (green color).

This method completes the progress bar with a success indicator, changing the color to green and optionally adding a success message.

Parameters:

Name	Type	Description	Default
desc	Optional[str]	Success message to display	None
close	bool	Whether to close the progress bar (default: True)	True
append_desc	bool	Whether to append to existing description (default: True)	True

Example

pbar = ProgressBar(total=100, desc="Processing")
for i in range(100):
    pbar.update(1)
pbar.success("All done!")  # Shows green bar with message

Source code in src/concurry/utils/progress.py

def success(self, desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None:
    """Mark the progress bar as successful (green color).

    This method completes the progress bar with a success indicator, changing
    the color to green and optionally adding a success message.

    Args:
        desc: Success message to display
        close: Whether to close the progress bar (default: True)
        append_desc: Whether to append to existing description (default: True)

    Example:
        ```python
        pbar = ProgressBar(total=100, desc="Processing")
        for i in range(100):
            pbar.update(1)
        pbar.success("All done!")  # Shows green bar with message
        ```
    """
    self._complete_with_status(
        color="#43a047",  # Dark Green
        desc=desc,
        close=close,
        append_desc=append_desc,
    )

stop(desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None

Mark the progress bar as stopped (grey color).

This method completes the progress bar with a stop indicator, changing the color to grey and optionally adding a stop message. Useful for indicating early termination or cancellation.

Parameters:

Name	Type	Description	Default
desc	Optional[str]	Stop message to display	None
close	bool	Whether to close the progress bar (default: True)	True
append_desc	bool	Whether to append to existing description (default: True)	True

Example

pbar = ProgressBar(total=100, desc="Processing")
for i in range(100):
    if should_stop():
        pbar.stop("Cancelled by user")  # Shows grey bar
        break
    pbar.update(1)

Source code in src/concurry/utils/progress.py

def stop(self, desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None:
    """Mark the progress bar as stopped (grey color).

    This method completes the progress bar with a stop indicator, changing
    the color to grey and optionally adding a stop message. Useful for
    indicating early termination or cancellation.

    Args:
        desc: Stop message to display
        close: Whether to close the progress bar (default: True)
        append_desc: Whether to append to existing description (default: True)

    Example:
        ```python
        pbar = ProgressBar(total=100, desc="Processing")
        for i in range(100):
            if should_stop():
                pbar.stop("Cancelled by user")  # Shows grey bar
                break
            pbar.update(1)
        ```
    """
    self._complete_with_status(
        color="#b0bec5",  # Dark Grey
        desc=desc,
        close=close,
        append_desc=append_desc,
    )

failure(desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None

Mark the progress bar as failed (red color).

This method completes the progress bar with a failure indicator, changing the color to red and optionally adding a failure message. Use in exception handlers to indicate errors.

Parameters:

Name	Type	Description	Default
desc	Optional[str]	Failure message to display	None
close	bool	Whether to close the progress bar (default: True)	True
append_desc	bool	Whether to append to existing description (default: True)	True

Example

pbar = ProgressBar(total=100, desc="Processing")
try:
    for i in range(100):
        if error_occurred():
            raise ValueError("Error")
        pbar.update(1)
    pbar.success()
except Exception as e:
    pbar.failure(f"Failed: {e}")  # Shows red bar
    raise

Source code in src/concurry/utils/progress.py

def failure(self, desc: Optional[str] = None, close: bool = True, append_desc: bool = True) -> None:
    """Mark the progress bar as failed (red color).

    This method completes the progress bar with a failure indicator, changing
    the color to red and optionally adding a failure message. Use in exception
    handlers to indicate errors.

    Args:
        desc: Failure message to display
        close: Whether to close the progress bar (default: True)
        append_desc: Whether to append to existing description (default: True)

    Example:
        ```python
        pbar = ProgressBar(total=100, desc="Processing")
        try:
            for i in range(100):
                if error_occurred():
                    raise ValueError("Error")
                pbar.update(1)
            pbar.success()
        except Exception as e:
            pbar.failure(f"Failed: {e}")  # Shows red bar
            raise
        ```
    """
    self._complete_with_status(
        color="#e64a19",  # Dark Red
        desc=desc,
        close=close,
        append_desc=append_desc,
    )

refresh() -> None

Refresh the progress bar display.

This method manually updates the progress bar display. It's typically called automatically by other methods, but can be called manually if needed.

Example

pbar = ProgressBar(total=100)
pbar.color = "#9c27b0"  # Change color
pbar.refresh()  # Apply the change

Source code in src/concurry/utils/progress.py

def refresh(self) -> None:
    """Refresh the progress bar display.

    This method manually updates the progress bar display. It's typically
    called automatically by other methods, but can be called manually if needed.

    Example:
        ```python
        pbar = ProgressBar(total=100)
        pbar.color = "#9c27b0"  # Change color
        pbar.refresh()  # Apply the change
        ```
    """
    self.pbar.colour = self.color
    self.pbar.refresh()

close() -> None

Close and clean up the progress bar.

This method closes the progress bar and releases any resources. It's automatically called by success(), failure(), and stop() when close=True.

Example

pbar = ProgressBar(total=100)
try:
    for i in range(100):
        pbar.update(1)
finally:
    pbar.close()  # Ensure cleanup

Source code in src/concurry/utils/progress.py

def close(self) -> None:
    """Close and clean up the progress bar.

    This method closes the progress bar and releases any resources. It's
    automatically called by success(), failure(), and stop() when close=True.

    Example:
        ```python
        pbar = ProgressBar(total=100)
        try:
            for i in range(100):
                pbar.update(1)
        finally:
            pbar.close()  # Ensure cleanup
        ```
    """
    self.pbar.refresh()
    self.pbar.close()
    self.pbar.refresh()

---

Usage Examples

Basic Usage

from concurry.utils.progress import ProgressBar
import time

# Wrap an iterable
for item in ProgressBar(range(100), desc="Processing"):
    time.sleep(0.01)
# Automatically shows success!

# Manual progress bar
pbar = ProgressBar(total=100, desc="Manual")
for i in range(100):
    time.sleep(0.01)
    pbar.update(1)
pbar.success("Complete!")

Customization

from concurry.utils.progress import ProgressBar

# Custom colors and settings
pbar = ProgressBar(
    total=100,
    desc="Processing",
    unit="files",
    color="#9c27b0",  # Purple
    ncols=120,
    smoothing=0.1,
    miniters=5
)

for i in range(100):
    pbar.update(1)

pbar.success()

State Management

from concurry.utils.progress import ProgressBar

pbar = ProgressBar(total=100, desc="Processing")

try:
    for i in range(100):
        if error_condition:
            raise ValueError("Error occurred")
        process(i)
        pbar.update(1)
    pbar.success("All done!")  # Green
except Exception as e:
    pbar.failure(f"Failed: {e}")  # Red
    raise

Dynamic Updates

from concurry.utils.progress import ProgressBar

pbar = ProgressBar(total=100, desc="Phase 1")

for i in range(100):
    # Update description dynamically
    if i == 50:
        pbar.set_description("Phase 2")

    # Update total if needed
    if i == 75 and more_work_discovered:
        pbar.set_total(150)

    pbar.update(1)

pbar.success()

Constructor Parameters

Required Parameters

• total (int, optional): Total number of items to process. Required for manual progress bars.

Optional Parameters

Display Options

• desc (str): Description shown before the progress bar
• unit (str): Unit name for items being processed (default: "row")
• color (str): Hex color code for the progress bar (default: "#0288d1")
• ncols (int): Width of the progress bar in characters (default: 100)

Behavior Options

• style (Literal["auto", "notebook", "std", "ray"]): Progress bar style (default: "auto")
• smoothing (float): Smoothing factor for progress updates (default: 0.15)
• disable (bool): Whether to disable the progress bar (default: False)
• miniters (int): Minimum iterations between updates (default: 1)

Advanced Options

• pbar (TqdmProgressBar, optional): Existing tqdm progress bar instance
• progress_bar (Union[bool, dict, ProgressBar]): Progress bar configuration or instance
• prefer_kwargs (bool): Whether to prefer kwargs over progress_bar dict (default: True)

Methods

update()

Update the progress bar by n steps.

pbar.update(n=1)  # Increment by 1
pbar.update(10)   # Increment by 10

Parameters: - n (int): Number of steps to increment (default: 1)

Returns: - Optional[bool]: True if display was updated, None if buffered

---

set_n()

Set the current progress value directly.

pbar.set_n(50)  # Set to 50

Parameters: - new_n (int): New progress value

---

set_total()

Set the total number of items.

pbar.set_total(200)  # Change total to 200

Parameters: - new_total (int): New total value

---

set_description()

Set the description text.

pbar.set_description("Phase 2")
pbar.set_description("Processing files", refresh=True)

Parameters: - desc (str, optional): New description - refresh (bool, optional): Whether to refresh display (default: True)

Returns: - Optional[str]: The previous description

---

set_unit()

Set the unit name.

pbar.set_unit("files")
pbar.set_unit("MB")

Parameters: - new_unit (str): New unit name

---

success()

Mark the progress bar as successful (green color).

pbar.success()  # Default success message
pbar.success("All done!")  # Custom message
pbar.success("Complete", close=False)  # Keep open

Parameters: - desc (str, optional): Success message - close (bool): Whether to close the progress bar (default: True) - append_desc (bool): Whether to append to existing description (default: True)

---

failure()

Mark the progress bar as failed (red color).

pbar.failure()  # Default failure message
pbar.failure("Error occurred!")  # Custom message

Parameters: - desc (str, optional): Failure message - close (bool): Whether to close the progress bar (default: True) - append_desc (bool): Whether to append to existing description (default: True)

---

stop()

Mark the progress bar as stopped (grey color).

pbar.stop()  # Default stop message
pbar.stop("Cancelled by user")  # Custom message

Parameters: - desc (str, optional): Stop message - close (bool): Whether to close the progress bar (default: True) - append_desc (bool): Whether to append to existing description (default: True)

---

refresh()

Manually refresh the progress bar display.

pbar.refresh()

---

close()

Close and clean up the progress bar.

pbar.close()

Type Signatures

ProgressBar Class

from concurry.utils.progress import ProgressBar
from typing import Literal, Optional, Union

class ProgressBar:
    def __init__(
        self,
        *args,
        pbar: Optional[TqdmProgressBar] = None,
        style: Literal["auto", "notebook", "std", "ray"] = "auto",
        unit: str = "row",
        color: str = "#0288d1",
        ncols: int = 100,
        smoothing: float = 0.15,
        total: Optional[int] = None,
        disable: bool = False,
        miniters: int = 1,
        progress_bar: Union[bool, dict, ProgressBar] = True,
        prefer_kwargs: bool = True,
        **kwargs,
    ) -> None:
        ...

    def update(self, n: int = 1) -> Optional[bool]:
        ...

    def set_n(self, new_n: int) -> None:
        ...

    def set_total(self, new_total: int) -> None:
        ...

    def set_description(
        self,
        desc: Optional[str] = None,
        refresh: Optional[bool] = True
    ) -> Optional[str]:
        ...

    def set_unit(self, new_unit: str) -> None:
        ...

    def success(
        self,
        desc: Optional[str] = None,
        close: bool = True,
        append_desc: bool = True
    ) -> None:
        ...

    def stop(
        self,
        desc: Optional[str] = None,
        close: bool = True,
        append_desc: bool = True
    ) -> None:
        ...

    def failure(
        self,
        desc: Optional[str] = None,
        close: bool = True,
        append_desc: bool = True
    ) -> None:
        ...

    def refresh(self) -> None:
        ...

    def close(self) -> None:
        ...

Progress Bar Styles

auto

Automatically detects the environment: - Uses notebook style in Jupyter/IPython - Uses standard terminal style otherwise

notebook

Optimized for Jupyter notebooks with rich HTML widgets.

std

Standard terminal-based progress bar.

ray

Integrates with Ray's distributed progress tracking (requires Ray).

Color Codes

Default colors for different states:

• Progress: #0288d1 (blue)
• Success: #43a047 (green)
• Failure: #e64a19 (red)
• Stop: #b0bec5 (grey)

Custom colors can be specified as hex codes:

pbar = ProgressBar(total=100, color="#9c27b0")  # Purple

See Also

• Progress User Guide - Learn how to use progress bars
• Examples - See practical examples
• Futures API - Futures API reference