Add run_lanes(func, lanes) and resolve_jobs(ctx, count)

Author: kdeldycke

changelog.md

@@ -13,6 +13,9 @@
 - Add a `--theme=auto` value that selects the `dark` or `light` palette from the detected terminal background. Accepted on every `@command` and `@group`; the `--theme` default stays `dark`.
 - Detect the terminal background from the `CLITHEME` and `COLORFGBG` environment variables, and, when a `ThemeOption` is built with `query_background=True`, from a live xterm OSC 11 terminal query.
 - Add `resolve_background` and `query_osc_background` to `click_extra.color`, and `resolve_auto_theme` and the `AUTO_THEME` constant to `click_extra.theme`.
+- Add `run_lanes(func, lanes)` to `click_extra.execution`: a fan-out that runs each lane's items serially while running distinct lanes concurrently, sized by the resolved `--jobs` count. `run_jobs` is its single-item-per-lane special case.
+- Add `resolve_jobs(ctx, count)` to `click_extra.execution`, exposing the worker-count policy shared by `run_jobs` and `run_lanes` for callers that need the resolved count before fanning out. Both helpers gain a `serial_at_debug` keyword that collapses to sequential at `DEBUG` verbosity.
+- Publish `format_cli_prompt` in `click_extra.testing` (was the private `_format_cli_prompt`): render a themed, copy-pasteable prompt simulating a CLI invocation.
 
 ## [`8.1.4` (2026-06-27)](https://github.com/kdeldycke/click-extra/compare/v8.1.3...v8.1.4)
 

click_extra/__init__.py

@@ -150,7 +150,9 @@
     JobsOption,
     TimerOption,
     ZeroExitOption,
+    resolve_jobs,
     run_jobs,
+    run_lanes,
 )
 from .highlight import (
     HelpFormatter,
@@ -217,7 +219,7 @@
     parse_test_suite,
     run_test_suite,
 )
-from .testing import CliRunner, Result
+from .testing import CliRunner, Result, format_cli_prompt
 from .theme import (
     BUILTIN_THEMES,
     HelpTheme,
@@ -355,6 +357,7 @@
     "edit",
     "file_path",
     "flatten_config_keys",
+    "format_cli_prompt",
     "format_filename",
     "format_from_path",
     "format_param_row",
@@ -404,8 +407,10 @@
     "render_manpages",
     "render_table",
     "require_sibling_param",
+    "resolve_jobs",
     "run_config_validation",
     "run_jobs",
+    "run_lanes",
     "run_test_suite",
     "search_params",
     "secho",

click_extra/execution.py

@@ -279,20 +279,64 @@ def __init__(
         )
 
 
+def resolve_jobs(
+    ctx: click.Context | None,
+    count: int,
+    *,
+    serial_at_debug: bool = False,
+) -> int:
+    """Resolve how many worker threads to use for a batch of ``count`` items.
+
+    Returns the number of items to process in parallel; ``1`` means run
+    sequentially in the calling thread. This is the policy shared by
+    :func:`run_jobs` and :func:`run_lanes`, exposed on its own for callers that
+    must know the resolved count *before* they fan out (for example to pick a
+    progress-rendering mode). It collapses to sequential when:
+
+    - there is no active CLI context (programmatic or test use),
+    - a single item leaves nothing to parallelize, or
+    - the resolved :class:`JobsOption` count
+      (``ctx.meta[click_extra.context.JOBS]``) is ``1`` or less.
+
+    Otherwise that count wins, capped at ``count``: there is no point spinning up
+    more workers than there are items.
+
+    :param ctx: the active Click context, read for the resolved ``--jobs`` count
+        (and, with ``serial_at_debug``, the verbosity). ``None`` forces sequential.
+    :param count: how many items are about to be scheduled.
+    :param serial_at_debug: when set, also collapse to sequential at ``DEBUG``
+        verbosity, where coherent per-worker log narration matters more than the
+        speed-up (interleaved threads would scramble it). Off by default.
+    """
+    if count <= 1 or ctx is None:
+        return 1
+    # Compared against the stdlib level rather than click_extra.logging.LogLevel
+    # (which mirrors it) to keep this module free of a logging-module import cycle.
+    if serial_at_debug and context.get(ctx, context.VERBOSITY_LEVEL) == logging.DEBUG:
+        return 1
+    jobs = context.get(ctx, context.JOBS, 1)
+    return min(jobs, count) if jobs > 1 else 1
+
+
 def run_jobs(
     func: Callable[[T], R],
     items: Iterable[T],
     *,
     jobs: int | None = None,
+    serial_at_debug: bool = False,
 ) -> Iterator[R]:
     """Run ``func`` over ``items``, parallelized per the resolved ``--jobs`` count.
 
-    The worker count is taken from ``jobs`` when given, else from the active
-    command's :class:`JobsOption` value (``ctx.meta[click_extra.context.JOBS]``),
-    else ``1``. With a single worker (or at most one item) the items run
-    **sequentially and lazily**, so a caller can stop early on the first result
-    (for example to abort on the first failure); otherwise they run in a thread
-    pool. Either way results are yielded in submission order, like :func:`map`.
+    The worker count is taken from ``jobs`` when given, else resolved from the
+    active command's :class:`JobsOption` value by :func:`resolve_jobs`, else ``1``.
+    With a single worker (or at most one item) the items run **sequentially and
+    lazily**, so a caller can stop early on the first result (for example to abort
+    on the first failure); otherwise they run in a thread pool. Either way results
+    are yielded in submission order, like :func:`map`.
+
+    This is the single-task-per-item special case of :func:`run_lanes` (every item
+    is its own lane). Reach for :func:`run_lanes` when some items must run serially
+    relative to one another while others run concurrently.
 
     The pool is thread-based, which suits the I/O- and subprocess-bound work CLI
     tools usually parallelize (each child releases the GIL). The count is a
@@ -302,13 +346,15 @@ def run_jobs(
     :param items: The work items. Materialized up front to size the pool.
     :param jobs: Override the worker count instead of reading it from the
         context. ``1`` or fewer forces sequential execution.
+    :param serial_at_debug: forwarded to :func:`resolve_jobs` when ``jobs`` is not
+        given: collapse to sequential at ``DEBUG`` verbosity.
     :return: An iterator over ``func``'s results, in the order of ``items``.
     """
+    work = list(items)
     if jobs is None:
         ctx = click.get_current_context(silent=True)
-        jobs = context.get(ctx, context.JOBS, 1) if ctx is not None else 1
+        jobs = resolve_jobs(ctx, len(work), serial_at_debug=serial_at_debug)
 
-    work = list(items)
     if jobs <= 1 or len(work) <= 1:
         # Sequential and lazy: the caller can break early (for example on the
         # first failure) and the remaining items never run.
@@ -321,6 +367,65 @@ def run_jobs(
             yield from executor.map(func, work)
 
 
+def run_lanes(
+    func: Callable[[T], R],
+    lanes: Iterable[Iterable[T]],
+    *,
+    jobs: int | None = None,
+    serial_at_debug: bool = False,
+) -> Iterator[R]:
+    """Run ``func`` over grouped items: serial within a lane, concurrent across.
+
+    Each *lane* is an iterable of items. ``func`` is mapped over every item, but a
+    lane's own items run **serially and in order** on a single worker, while distinct
+    lanes run **concurrently** up to the resolved ``--jobs`` count. This is the right
+    primitive when some work must be serialized relative to itself (a shared lock, a
+    rate limit, one mailbox file, one package-manager backend) yet still overlap with
+    unrelated work.
+
+    :func:`run_jobs` is the degenerate case where every lane holds a single item.
+    Concurrency is sized by the *number of lanes* (one worker per lane), since a
+    lane never splits across workers.
+
+    Results are yielded in lane-submission order, a lane's items in order, like
+    :func:`map`. With a single worker the run stays lazy (the caller can break
+    early); otherwise every lane is submitted up front. A lane runs entirely on one
+    worker, so a stateful resource bound to the lane (a per-lane cache, a connection)
+    is touched by only that one thread and needs no lock.
+
+    :param func: Called once per item; its return value is yielded.
+    :param lanes: The lanes, each an iterable of items. Materialized up front.
+    :param jobs: Override the worker count instead of reading it from the context.
+        ``1`` or fewer forces fully sequential execution.
+    :param serial_at_debug: forwarded to :func:`resolve_jobs` when ``jobs`` is not
+        given: collapse to sequential at ``DEBUG`` verbosity.
+    :return: An iterator over ``func``'s results, lane by lane in submission order.
+    """
+    lane_list = [list(lane) for lane in lanes]
+    if not lane_list:
+        return
+    if jobs is None:
+        ctx = click.get_current_context(silent=True)
+        jobs = resolve_jobs(ctx, len(lane_list), serial_at_debug=serial_at_debug)
+    elif jobs > 1:
+        jobs = min(jobs, len(lane_list))
+
+    if jobs <= 1:
+        # Sequential and lazy across every lane and item: the caller can break early.
+        for lane in lane_list:
+            for item in lane:
+                yield func(item)
+    else:
+        # Each lane is a serial chain run on one worker; chains run concurrently and
+        # their results are yielded in submission order.
+        def run_chain(lane: list[T]) -> list[R]:
+            return [func(item) for item in lane]
+
+        with ThreadPoolExecutor(max_workers=jobs) as executor:
+            for chain_results in executor.map(run_chain, lane_list):
+                yield from chain_results
+
+
 class TimerOption(ExtraOption):
     """A pre-configured option that is adding a ``--time``/``--no-time`` flag to print
     elapsed time at the end of CLI execution.

click_extra/testing.py

@@ -162,11 +162,17 @@ def args_cleanup(*args: TArg | TNestedArgs) -> tuple[str, ...]:
     return tuple(str(arg) for arg in flatten(args) if arg is not None)
 
 
-def _format_cli_prompt(
+def format_cli_prompt(
     cmd_args: Iterable[str],
     extra_env: TEnvVars | None = None,
 ) -> str:
-    """Simulate the console prompt used to invoke the CLI."""
+    """Render the shell prompt simulating a CLI invocation, for logs and dry-runs.
+
+    Prefixes :data:`PROMPT` to any ``extra_env`` assignments and the command line,
+    each styled through the active theme
+    (:func:`~click_extra.theme.get_current_theme`). Useful to print a
+    copy-pasteable command trace in debug logs, dry-runs and test output.
+    """
     active_theme = get_current_theme()
     extra_env_string = ""
     if extra_env:
@@ -188,7 +194,7 @@ def render_cli_run(
 
     Mostly used to print debug traces to user or in test results.
     """
-    prompt = _format_cli_prompt(args, env)
+    prompt = format_cli_prompt(args, env)
 
     if isinstance(result, click.testing.Result):
         view = StreamView.from_result(result)

docs/execution.md

@@ -147,6 +147,35 @@ assert "Baked APPLE" in result.stdout
 
 The pool is thread-based, which fits the I/O- and subprocess-bound work CLIs usually parallelize (each child releases the GIL). With a single worker the run stays lazy, so a caller can stop on the first result, for example to abort on the first failure.
 
+## Running lanes in parallel
+
+Sometimes work cannot all run concurrently: a subset must be serialized relative to itself (a shared lock, a rate limit, one mailbox file read at a time, one package-manager backend) while still overlapping with unrelated subsets. `run_lanes(func, lanes)` groups items into *lanes*: a lane's own items run serially and in order on a single worker, while distinct lanes run concurrently up to the resolved `--jobs` count. `run_jobs` is the degenerate case where every lane holds a single item.
+
+```{click:source}
+from click import command, echo
+from click_extra import jobs_option, run_lanes
+
+@command
+@jobs_option
+def bake():
+    """Bake several trays: items on a tray bake in order, trays bake in parallel."""
+    trays = (("apple", "banana"), ("cherry",))
+    for baked in run_lanes(str.upper, trays):
+        echo(f"Baked {baked}")
+```
+
+```{click:run}
+result = invoke(bake, args=["--jobs", "2"])
+assert result.exit_code == 0
+assert "Baked APPLE" in result.stdout
+```
+
+Concurrency is sized by the number of lanes (one worker per lane), and results are yielded in lane-submission order. Because a lane runs entirely on one worker, a stateful resource bound to that lane (a per-lane cache, a connection) is touched by only one thread and needs no lock.
+
+## Resolving the job count
+
+`run_jobs` and `run_lanes` decide their worker count internally, but a caller that must know it *before* fanning out (for example to pick a progress-rendering mode) can call `resolve_jobs(ctx, count)` directly. It returns the same number those helpers use: `1` (sequential) when there is no context, a single item, or `--jobs 1`, otherwise the resolved count capped at `count`. Passing `serial_at_debug=True` also collapses to sequential at `DEBUG` verbosity, where coherent per-worker log narration matters more than the speed-up; both helpers forward this flag.
+
 ## Zero exit code
 
 A pre-configured `-0`/`--zero-exit` option flag, following the convention popularized by linters and static analysers: they exit with a non-zero code whenever they report findings, so automation can gate on it. Setting this flag flips that behavior, so the CLI returns `0` as long as it ran to completion, reserving non-zero codes for actual execution failures.