python-script-runner 7.4.4

Production-grade Python script execution engine with comprehensive monitoring, alerting, analytics, and enterprise integrations

Python Script Runner

A production-grade Python script execution engine with comprehensive monitoring, alerting, analytics, and real-time visualization.

---

Features

• Script execution with timeout, retry, and environment management
• Real-time visualization of the full execution pipeline
• DAG-based workflow orchestration with parallel execution
• Metrics collection — CPU, memory, I/O, timing per run
• Alert management — rule-based triggers via Slack, email, webhooks
• History & trend analysis — SQLite persistence with anomaly detection
• CI/CD integration — JUnit XML, TAP output, performance gates
• Remote execution — SSH, Docker, Kubernetes
• Web dashboard — FastAPI backend with WebSocket live updates

---

Visualization

Run any script with real-time orchestration visualization using the --visualize flag:

python runner.py my_script.py --visualize

Each step of the pipeline is displayed with elapsed time and per-step duration (e.g. (0.101s)). Status symbols:

Symbol	Meaning
⏳	Running
✓	Done
⊘	Skipped
✗	Error
🚀	Subprocess launched

Output file

Write a clean (ANSI-free) copy to disk:

python runner.py my_script.py --visualize --visualize-output run.log

JSON output format

Machine-readable structured output for CI pipelines and integrations:

python runner.py my_script.py --visualize --visualize-format json

The JSON document contains a header, a steps list with per-step elapsed_s and duration_s, and a footer. Access it programmatically with get_execution_report():

from runner import ExecutionVisualizer

v = ExecutionVisualizer(enabled=True, output_format="json", output_file="run.log")
v.show_header("pipeline.py")
# ... steps recorded automatically during runner.run_script() ...
v.show_footer(1.23, success=True)

report = v.get_execution_report()
slow_steps = [s for s in report["steps"] if s.get("duration_s", 0) > 0.5]

---

Workflow Orchestration

Execute multiple scripts as a DAG with optional parallelism:

from runner import ScriptWorkflow

wf = ScriptWorkflow(
    name="data_pipeline",
    max_parallel=2,          # run up to 2 scripts concurrently
    stop_on_failure=True,    # abort if any script fails
    on_step_callback=lambda name, status, result: print(f"{name}: {status}"),
)

wf.add_script("fetch",     "fetch.py")
wf.add_script("transform", "transform.py", dependencies=["fetch"])
wf.add_script("validate",  "validate.py",  dependencies=["fetch"])
wf.add_script("load_db",   "load_db.py",   dependencies=["transform", "validate"])

# Visualize the DAG before running
print(wf.visualize_dag())

result = wf.execute()

visualize_dag()

Prints an ASCII-art dependency graph showing node names, dependency arrows, and live execution status:

Workflow: data_pipeline
─────────────────────────────────────────────
[fetch       ] (pending)
    └──▶ [transform   ] (pending)
        └──▶ [load_db     ] (pending)
    └──▶ [validate    ] (pending)
─────────────────────────────────────────────

execute() result

{
    "status": "completed",   # or "aborted" if stop_on_failure triggered
    "total_scripts": 4,
    "successful": 4,
    "failed": 0,
    "total_time": 0.054,
    "results": { ... }       # per-script exit codes, timings, success flags
}

---

CLI Reference

usage: runner.py [-h] [--timeout TIMEOUT] [--visualize]
                 [--visualize-format {text,json}]
                 [--visualize-output FILE]
                 [--retry N] [--retry-strategy {linear,exponential,fibonacci}]
                 [--monitor-interval SECONDS]
                 [--show-history] [--analyze-trend]
                 [--dashboard] ...
                 script [script_args ...]

Key flags:

Flag	Description
--visualize	Show real-time execution flow
--visualize-format {text,json}	Output format (default: text)
--visualize-output FILE	Also write visualization to a file
--retry N	Retry on failure up to N times
--retry-strategy	linear, exponential, fibonacci, exponential_jitter
--timeout SECONDS	Kill script after N seconds
--monitor-interval S	Metric sampling interval (default: 0.1s)
--show-history	Print recent execution history
--analyze-trend	Run trend analysis on metric history
--dashboard	Start the web dashboard

---

Installation

pip install python-script-runner

Or from source:

git clone https://github.com/jomardyan/Python-Script-Runner
cd Python-Script-Runner
pip install -e .

---

👨‍💻 Author

Hayk Jomardyan

• 🌐 Website: lolino.pl
• 📧 Email: [e-mail]
• 💼 GitHub: @jomardyan

License

MIT License - See LICENSE file for details