Run Commands from Python: Timeouts, Errors, and Log Files

A Python pipeline calls an external tool, the tool fails, and the pipeline continues with an incomplete output file. Or a command waits for input that nobody will provide. Both problems become harder to diagnose when the terminal output is gone.

This tutorial builds a small command runner that stops on errors, applies a timeout, and writes output to a fresh log. The demo uses Python itself as the child process, so you can test every case without installing another tool.

1. Define the runner’s job

This helper runs one trusted executable at a time. It uses an argument list, closes standard input, and sends both output streams to disk. Each run needs a new log filename. It does not retry commands or manage a job queue.

The key API behavior comes from Python’s subprocess.run documentation: check=True raises on a nonzero exit, and timeout kills and waits for the direct child before raising. Process creation can delay the timeout. An argument list handles spaces without manual shell quoting.

2. Download and run a successful command

Use Python 3.10 or newer. No external packages are required. The examples were tested on macOS with Python 3.14.7; the shell commands below assume a POSIX shell such as bash or zsh.

Download run_logged.py into a new working directory, then run:

python3 --version
python3 run_logged.py --help
mkdir subprocess-demo
python3 run_logged.py ok subprocess-demo/ok.log
cat subprocess-demo/ok.log

Expected output from the last two commands:

OK: log=subprocess-demo/ok.log
processed 3 records

Use fresh paths when repeating the examples. Existing logs are deliberately preserved.

3. The complete implementation

"""Run a trusted command with a fresh log and a timeout. Python 3.10+."""
import argparse
import math
from pathlib import Path
import subprocess
import sys


def run_logged(argv, log_path, *, timeout=60):
    """Keep stdout/stderr on disk; propagate launch, exit and timeout errors."""
    if isinstance(argv, (str, bytes)) or not argv:
        raise ValueError("argv must be a nonempty argument sequence")
    if not math.isfinite(timeout) or timeout <= 0:
        raise ValueError("timeout must be a finite positive number")
    # Exclusive creation prevents accidental replacement of an earlier log.
    with Path(log_path).open("xb") as log:
        return subprocess.run(
            argv,
            stdin=subprocess.DEVNULL,
            stdout=log,
            stderr=subprocess.STDOUT,
            shell=False,
            check=True,
            timeout=timeout,
        )


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("mode", choices=["ok", "fail", "timeout"])
    parser.add_argument("log", type=Path, help="new log path; parent must exist")
    args = parser.parse_args()
    programs = {
        "ok": "print('processed 3 records', flush=True)",
        "fail": "import sys; print('invalid input', file=sys.stderr, flush=True); sys.exit(2)",
        "timeout": "import time; print('started', flush=True); time.sleep(30)",
    }
    command = [sys.executable, "-u", "-c", programs[args.mode]]
    try:
        run_logged(command, args.log, timeout=1 if args.mode == "timeout" else 10)
    except subprocess.CalledProcessError as exc:
        print(f"FAILED: exit={exc.returncode}; log={args.log}", file=sys.stderr)
        return 1
    except subprocess.TimeoutExpired:
        print(f"TIMEOUT: log={args.log}", file=sys.stderr)
        return 1
    except (OSError, ValueError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    print(f"OK: log={args.log}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

The demo’s -c programs are fixed strings under our control. sys.executable selects the interpreter running the parent script. In your pipeline, replace that command list with the actual executable and its arguments.

Opening the log with xb rejects an existing path before starting the command. Binary mode preserves the child’s output bytes. The parent directory must exist. If launch fails after the log is opened, an empty log can remain; that file alone is not proof that a job ran.

4. Exercise failure and timeout paths

Run these commands one at a time. The two runner calls intentionally return exit status 1, so do not put this demonstration in a script that stops at its first error.

python3 run_logged.py fail subprocess-demo/fail.log
cat subprocess-demo/fail.log
python3 run_logged.py timeout subprocess-demo/timeout.log
cat subprocess-demo/timeout.log

The failed command produces:

FAILED: exit=2; log=subprocess-demo/fail.log
invalid input

The timeout produces:

TIMEOUT: log=subprocess-demo/timeout.log
started

exit=2 is the child’s status; the demo CLI maps both failure and timeout to its own status 1. Importing run_logged directly preserves the original exceptions for your application to handle.

The timeout demo uses unbuffered Python output and flushes its first message. Other programs may buffer their output, leaving a partial or empty log when stopped. Logging to a file also means output consumes disk space: set retention and size policies for long-running jobs.

5. Pass a filename containing spaces

Save the following as count_demo.py beside run_logged.py, in a scratch directory. It creates a tiny worker script and input file, then counts nonempty lines. These fixture files are overwritten if they already exist.

import sys
from pathlib import Path
from run_logged import run_logged

worker = Path("count_records.py")
worker.write_text(
    "import sys\n"
    "from pathlib import Path\n"
    "with Path(sys.argv[1]).open(encoding='utf-8') as handle:\n"
    "    print(sum(1 for line in handle if line.strip()))\n",
    encoding="utf-8",
)
Path("sample records.txt").write_text("alpha\nbeta\ngamma\n", encoding="utf-8")
run_logged(
    [sys.executable, str(worker.resolve()), "sample records.txt"],
    "count.log",
    timeout=10,
)
assert Path("count.log").read_text().strip() == "3"
print("PASS: counted 3 nonempty lines")

Run python3 count_demo.py. It prints PASS: counted 3 nonempty lines. The filename is one list element; adding literal quote characters around it would change the argument.

For a real pipeline, validate the output before recording completion. Pair that check with the atomic JSON save helper when updating a small progress file, and keep an input checksum manifest if you need to identify the exact data used.

6. Check that old logs survive

After the counting example, save this as check_log.py and run python3 check_log.py in the same directory:

from pathlib import Path
from run_logged import run_logged
import sys

path = Path("count.log")
before = path.read_bytes()
try:
    run_logged([sys.executable, "-c", "print('replacement')"], path)
except FileExistsError:
    assert path.read_bytes() == before
    print("PASS: existing log preserved")
else:
    raise AssertionError("Expected FileExistsError")

This verifies that a repeated run cannot silently erase the previous log. For production runs, choose a new run directory or unique log name rather than deleting logs automatically.

7. Troubleshooting and boundaries

The runner does not cap log size, roll back partial result files, or guarantee termination of descendants spawned by the direct child. Tools that launch worker processes need a process-tree strategy appropriate to the operating system or a job supervisor. A zero exit status still needs an application-level output check.

Keep commands and options trusted. shell=False is not a sandbox, and a tool can still interpret an argument as an option. Windows batch files also need special care; see Python’s subprocess security notes. This example targets ordinary executables and Python scripts.

8. Summary

Give each command a fresh log, a deliberate timeout, and an explicit error path. Test failure cases before inserting the runner into a batch pipeline. Treat process success and valid output as separate checks before saving progress.