"""`run()`: execute a shell command per item, in parallel, with a bar.
A progress-bar'd ``xargs -P`` in Python. Templates never go through
`str.format` -- only the exact placeholder tokens ``{}`` and ``{item}``
are substituted -- so commands containing literal braces (``awk
'{print $1}'``) pass through untouched.
"""
from __future__ import annotations
import functools
import os
import shlex
import subprocess
import typing
from . import _sync
_PLACEHOLDERS: tuple[str, str] = ('{}', '{item}')
CommandT = (
str
| typing.Sequence[str]
| typing.Callable[[typing.Any], typing.Sequence[str]]
)
def _substitute(token: str, item_text: str) -> str:
"""Replace the placeholder spellings inside one token."""
for placeholder in _PLACEHOLDERS:
token = token.replace(placeholder, item_text)
return token
def _has_placeholder(tokens: typing.Iterable[str]) -> bool:
"""Return whether any token contains a placeholder."""
return any(
placeholder in token
for token in tokens
for placeholder in _PLACEHOLDERS
)
def build_argv(
command: CommandT, item: typing.Any, *, shell: bool
) -> list[str] | str:
"""Build the command for one item from a template.
Args:
command: A str template (split with `shlex.split`, non-POSIX
mode on Windows so backslash paths survive), an argv list
template, or a callable returning the argv. In the str and
list forms every ``{}``/``{item}`` inside a token is
replaced by ``str(item)`` -- an item containing spaces
stays a single argv element. Without any placeholder the
item is appended as the final argument (the ``xargs``
convention).
item: The batch item; substituted as ``str(item)``.
shell: With the str form, substitute into (and return) the
whole command string for ``subprocess.run(shell=True)``;
an appended item is `shlex.quote`-escaped. The caller must
trust its items -- see `run`.
Returns:
The argv list, or the command string when ``shell=True``.
"""
item_text: str = str(item)
if callable(command):
return [str(part) for part in command(item)]
if isinstance(command, str):
if shell:
if _has_placeholder((command,)):
return _substitute(command, item_text)
return f'{command} {shlex.quote(item_text)}'
tokens: list[str] = shlex.split(command, posix=os.name != 'nt')
else:
tokens = list(command)
if _has_placeholder(tokens):
return [_substitute(token, item_text) for token in tokens]
return [*tokens, item_text]
def _run_one(
command: CommandT,
item: typing.Any,
*,
check: bool,
capture_output: bool,
text: bool,
shell: bool,
cwd: typing.Any,
env: typing.Any,
) -> subprocess.CompletedProcess[typing.Any]:
"""Execute the command for one item (thread-pool worker)."""
argv: list[str] | str = build_argv(command, item, shell=shell)
return subprocess.run(
argv,
check=check,
capture_output=capture_output,
text=text,
shell=shell,
cwd=cwd,
env=env,
)
def make_runner(
command: CommandT,
*,
check: bool = True,
capture_output: bool = True,
text: bool = True,
shell: bool = False,
cwd: typing.Any = None,
env: typing.Any = None,
) -> typing.Callable[[typing.Any], subprocess.CompletedProcess[typing.Any]]:
"""Bind a command template into a per-item callable for `map`."""
return functools.partial(
_run_one,
command,
check=check,
capture_output=capture_output,
text=text,
shell=shell,
cwd=cwd,
env=env,
)
def run(
command: CommandT,
items: typing.Iterable[typing.Any],
/,
*,
check: bool = True,
capture_output: bool = True,
text: bool = True,
shell: bool = False,
cwd: typing.Any = None,
env: typing.Any = None,
**kwargs: typing.Any,
) -> list[subprocess.CompletedProcess[typing.Any]]:
"""Run a shell command for every item in parallel, with a bar.
``progressbar.run('gzip -k {}', files, workers=4)`` is a
progress-bar'd ``xargs -P``. Subprocesses release the GIL, so this
always runs on threads (`Pool.run` reuses a pool's executor).
Args:
command: Template -- see `build_argv` for the three forms and
the placeholder rules.
items: The batch; each becomes one subprocess.
check: Raise `subprocess.CalledProcessError` on a non-zero
exit (feeding `on_error` like any other worker error).
capture_output: Capture stdout/stderr into the results --
the default, so child output cannot corrupt the bar.
text: Decode captured output as text.
shell: Run through the shell (str form only). The items are
substituted into the command line: only use with trusted
items, this is the documented injection risk.
cwd: Working directory for the subprocesses.
env: Environment for the subprocesses.
**kwargs: The shared execution keywords (`workers`, `bar`,
`on_error`, `timeout`, ...); see `_sync.execute`.
Returns:
One `subprocess.CompletedProcess` per item, in input order
(exceptions in place under ``on_error='return'``).
"""
if 'pool' in kwargs:
raise TypeError(
'run() always uses threads (subprocesses release the GIL); '
'use Pool.run() to reuse an existing pool'
)
runner = make_runner(
command,
check=check,
capture_output=capture_output,
text=text,
shell=shell,
cwd=cwd,
env=env,
)
return _sync.map(runner, items, pool='thread', **kwargs)