Shell

Run shell commands.

Functions⚓︎

capture_shell ⚓︎

capture_shell(cmd, *, timeout=120, cwd=None, printer=None, validate_cmd=False)

Run shell command, return the output, and optionally print in real time.

WARNING: This function uses shell=True which can be a security risk. Only use with trusted input or enable validate_cmd for basic protection.

Inspired by: https://stackoverflow.com/a/38745040/3219667

PARAMETER	DESCRIPTION
`cmd`	shell command TYPE: `str`
`timeout`	process timeout in seconds. Defaults to 2 minutes. Use None for no timeout. TYPE: `int \| None` DEFAULT: `120`
`cwd`	optional path for shell execution TYPE: `Path \| None` DEFAULT: `None`
`printer`	optional callable to output the lines in real time TYPE: `Callable[[str], None] \| None` DEFAULT: `None`
`validate_cmd`	if True, validates command for dangerous patterns. Default False to preserve backward compatibility and allow legitimate shell features. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`str`	stripped output TYPE: `str`

RAISES	DESCRIPTION
`CalledProcessError`	if return code is non-zero
`TimeoutExpired`	if timeout is reached
`ValueError`	if validate_cmd=True and dangerous patterns are detected

Source code in corallium/shell.py

def capture_shell(
    cmd: str,
    *,
    timeout: int | None = 120,
    cwd: Path | None = None,
    printer: Callable[[str], None] | None = None,
    validate_cmd: bool = False,
) -> str:
    """Run shell command, return the output, and optionally print in real time.

    WARNING: This function uses shell=True which can be a security risk.
    Only use with trusted input or enable validate_cmd for basic protection.

    Inspired by: https://stackoverflow.com/a/38745040/3219667

    Args:
        cmd: shell command
        timeout: process timeout in seconds. Defaults to 2 minutes. Use None for no timeout.
        cwd: optional path for shell execution
        printer: optional callable to output the lines in real time
        validate_cmd: if True, validates command for dangerous patterns. Default False
            to preserve backward compatibility and allow legitimate shell features.

    Returns:
        str: stripped output

    Raises:
        CalledProcessError: if return code is non-zero
        TimeoutExpired: if timeout is reached
        ValueError: if validate_cmd=True and dangerous patterns are detected

    """
    if validate_cmd:
        _validate_shell_command(cmd)

    LOGGER.debug('Running', cmd=cmd, timeout=timeout, cwd=cwd, printer=printer, validate_cmd=validate_cmd)
    if timeout and timeout < 0:
        raise ValueError('Negative timeouts are not allowed')

    start = time()
    lines = []
    with subprocess.Popen(
        cmd,
        cwd=cwd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        universal_newlines=True,
        shell=True,
    ) as proc:
        if not (stdout := proc.stdout):
            raise NotImplementedError('Failed to read stdout from process.')
        return_code = None
        while return_code is None:
            if timeout is not None and time() - start >= timeout:
                proc.kill()
                break
            if line := stdout.readline():
                lines.append(line)
                if printer:
                    printer(line.rstrip())
            else:
                return_code = proc.poll()

    output = ''.join(lines)
    if return_code is None:
        # Process was killed due to timeout
        raise subprocess.TimeoutExpired(cmd=cmd, timeout=float(timeout or 0), output=output)
    if return_code != 0:
        raise subprocess.CalledProcessError(returncode=return_code, cmd=cmd, output=output)

    duration = time() - start
    LOGGER.debug('Shell command completed', cmd=cmd, returncode=0, duration_seconds=round(duration, 2), cwd=cwd)

    return output

capture_shell_async `async` ⚓︎

capture_shell_async(cmd, *, timeout=120, cwd=None, validate_cmd=False)

Run a shell command asynchronously and return the output.

WARNING: This function uses shell=True which can be a security risk. Only use with trusted input or enable validate_cmd for basic protection.

print(asyncio.run(capture_shell_async('ls ~/.config')))

PARAMETER	DESCRIPTION
`cmd`	shell command TYPE: `str`
`timeout`	process timeout in seconds. Defaults to 2 minutes. Use None for no timeout. TYPE: `int \| None` DEFAULT: `120`
`cwd`	optional path for shell execution TYPE: `Path \| None` DEFAULT: `None`
`validate_cmd`	if True, validates command for dangerous patterns. Default False to preserve backward compatibility and allow legitimate shell features. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`str`	stripped output TYPE: `str`

Source code in corallium/shell.py

async def capture_shell_async(
    cmd: str,
    *,
    timeout: int | None = 120,
    cwd: Path | None = None,
    validate_cmd: bool = False,
) -> str:
    """Run a shell command asynchronously and return the output.

    WARNING: This function uses shell=True which can be a security risk.
    Only use with trusted input or enable validate_cmd for basic protection.

    ```py
    print(asyncio.run(capture_shell_async('ls ~/.config')))
    ```

    Args:
        cmd: shell command
        timeout: process timeout in seconds. Defaults to 2 minutes. Use None for no timeout.
        cwd: optional path for shell execution
        validate_cmd: if True, validates command for dangerous patterns. Default False
            to preserve backward compatibility and allow legitimate shell features.

    Returns:
        str: stripped output

    """
    if validate_cmd:
        _validate_shell_command(cmd)

    LOGGER.debug('Running', cmd=cmd, timeout=timeout, cwd=cwd, validate_cmd=validate_cmd)
    start = time()
    return await asyncio.wait_for(
        _capture_shell_async(cmd=cmd, cwd=cwd, start_time=start),
        timeout=timeout or None,
    )

run_shell ⚓︎

run_shell(cmd, *, timeout=120, cwd=None, validate_cmd=False)

Run a shell command without capturing the output.

WARNING: This function uses shell=True which can be a security risk. Only use with trusted input or enable validate_cmd for basic protection.

PARAMETER	DESCRIPTION
`cmd`	shell command TYPE: `str`
`timeout`	process timeout in seconds. Defaults to 2 minutes. Use None for no timeout. TYPE: `int \| None` DEFAULT: `120`
`cwd`	optional path for shell execution TYPE: `Path \| None` DEFAULT: `None`
`validate_cmd`	if True, validates command for dangerous patterns. Default False to preserve backward compatibility and allow legitimate shell features. TYPE: `bool` DEFAULT: `False`

Source code in corallium/shell.py

def run_shell(cmd: str, *, timeout: int | None = 120, cwd: Path | None = None, validate_cmd: bool = False) -> None:
    """Run a shell command without capturing the output.

    WARNING: This function uses shell=True which can be a security risk.
    Only use with trusted input or enable validate_cmd for basic protection.

    Args:
        cmd: shell command
        timeout: process timeout in seconds. Defaults to 2 minutes. Use None for no timeout.
        cwd: optional path for shell execution
        validate_cmd: if True, validates command for dangerous patterns. Default False
            to preserve backward compatibility and allow legitimate shell features.

    """
    if validate_cmd:
        _validate_shell_command(cmd)

    LOGGER.debug('Running', cmd=cmd, timeout=timeout, cwd=cwd, validate_cmd=validate_cmd)

    start = time()
    subprocess.run(
        cmd,
        timeout=timeout or None,
        cwd=cwd,
        stdout=sys.stdout,
        stderr=sys.stderr,
        check=True,
        shell=True,
    )

    duration = time() - start
    LOGGER.debug('Shell command completed', cmd=cmd, returncode=0, duration_seconds=round(duration, 2), cwd=cwd)

Shell

Functions⚓︎

capture_shell ⚓︎

capture_shell_async async ⚓︎

run_shell ⚓︎

capture_shell_async `async` ⚓︎