Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/alex-ber/AlexBerUtils/llms.txt

Use this file to discover all available pages before exploring further.

The alexber.utils.processinvokes module wraps subprocess.run() with real-time log streaming. Long-running subprocesses write their output line-by-line to a Python logger instead of buffering until completion.

​
Quick start

1

Call initConfig() once at startup

initConfig() is called automatically when the module is imported, so this step is only needed if you want to override the defaults.
import logging
from alexber.utils import processinvokes

processinvokes.initConfig(
    default_log_name='myapp.subprocess',
    default_log_level=logging.DEBUG,
)
2

Invoke a subprocess

from alexber.utils.processinvokes import run_sub_process

run_sub_process('python', '-m', 'pytest', 'tests/')
Each line of output produced by pytest is forwarded to the configured logger at INFO level.

​
run_sub_process(*args, **kwargs)

Runs a subprocess and streams its stdout and stderr to a logger. This is a higher-level wrapper around subprocess.run(). The subprocess is executed in a thread pool so its output can be consumed line-by-line without deadlocks. 
from alexber.utils.processinvokes import run_sub_process

# Simple invocation
run_sub_process('git', 'clone', 'https://example.com/repo.git')

# With subprocess keyword arguments (cwd, env, etc.)
run_sub_process(
    'make', 'build',
    kwargs={'cwd': '/workspace', 'env': {'PATH': '/usr/bin'}},
)

# Override the log pipe for this call only
run_sub_process(
    'npm', 'install',
    logPipe={
        'cls': 'alexber.utils.processinvokes.LogPipe',
        'kwargs': {
            'logName': 'myapp.npm',
            'logLevel': logging.WARNING,
        },
    },
)

​
Parameters

ParameterDescription
*argsPositional arguments forwarded as the command and its arguments to subprocess.run().
kwargsDict of keyword arguments forwarded to subprocess.run() (e.g. cwd, env).
logPipeDict controlling the pipe object. Accepts cls (class or dotted string, default LogPipe) and kwargs (passed to the pipe constructor, e.g. logName, logLevel).
logSubprocessDict controlling the subprocess coordinator. Accepts cls (class or dotted string, default LogSubProcessCall) and kwargs.

​
Default subprocess.run() settings

Unless overridden via kwargs, the following defaults are applied:
SettingValueEffect
stdoutlog pipeSubprocess output goes to the logger.
stderrsubprocess.STDOUTstderr is merged into stdout.
textTrueOutput is decoded using the system default encoding.
bufsize1Line-buffered — each newline flushes immediately.
checkTrueNon-zero exit codes raise subprocess.CalledProcessError.
Because check=True is set by default, a non-zero exit code raises subprocess.CalledProcessError. Wrap the call in GuardedWorkerException when using run_sub_process inside a multiprocessing worker, because CalledProcessError is not picklable.

​
initConfig(**kwargs)

Configures module-level defaults. Call this once from the main thread before using run_sub_process. 
from alexber.utils import processinvokes
import logging

processinvokes.initConfig(
    default_log_name='myapp.subprocess',   # logger name
    default_log_level=logging.DEBUG,       # log level for subprocess output
    default_logpipe_cls='alexber.utils.processinvokes.FilePipe',  # write to file instead
    executor={'max_workers': 2, 'thread_name_prefix': 'sub'},
)
ParameterDefaultDescription
default_log_name'alexber.utils.processinvokes'Logger name used by LogPipe.
default_log_levellogging.INFOLog level for subprocess output lines.
default_logpipe_clsLogPipeClass used to create the pipe. Can be a class or dotted string.
default_log_subprocess_clsLogSubProcessCallClass used to coordinate the subprocess call.
executor{max_workers: 1, thread_name_prefix: 'processinvokes'}Arguments forwarded to ThreadPoolExecutor.

​
Pipe classes

​
LogPipe

The default pipe. Routes each output line to a Python logger. 
# Configured via logPipe kwargs in run_sub_process, or via initConfig
logPipe={
    'kwargs': {
        'logName': 'myapp.build',
        'logLevel': logging.INFO,
    }
}

​
FilePipe

An alternative pipe that writes output lines to a file instead of a logger. 
run_sub_process(
    'make', 'dist',
    logPipe={
        'cls': 'alexber.utils.processinvokes.FilePipe',
        'kwargs': {'fileName': 'build.log'},
    },
)
You can provide your own pipe class by subclassing BasePipe and overriding processLine(self, line). Pass the class (or its dotted import path) as logPipe.cls.

Build docs developers (and LLMs) love

Get started for freeTalk to us