pyrelay-workflow 0.1.0

A lightweight, async-first Python workflow engine with dynamic branching and rich visual dashboarding.

⚡ pyrelay

pyrelay is a lightweight, high-performance, async-first Python workflow engine. It allows developers to model, execute, and monitor complex task pipelines with native pythonic flows (if/else branching), zero complex DSLs, live terminal dashboards, and interactive web visualizers.

---

✨ Features

• 🐍 Native Python Branching: Use standard Python if/else and loops. The execution graph is discovered dynamically at runtime.
• ⚡ Async-First Architecture: Built natively on top of Python asyncio. Executes independent tasks concurrently.
• 🧵 Auto-Offloading for Sync Tasks: Run blocking synchronous functions without stalling the async event loop (automatically offloaded to thread executors).
• ⏱️ Fault Tolerance: Custom task-level retry policies, exponential backoff, and timeouts out-of-the-box.
• 🖥️ Live Console Dashboard: Animated status trees, progress bars, and log streams in your terminal powered by rich.
• 🌐 Draggable HTML/JS Visualizer: Compile any workflow run into a standalone dark-mode HTML file containing a responsive network chart powered by Vis-Network.

---

📦 Installation

Initialize your project and install pyrelay:

pip install pyrelay

(Requires Python 3.9+)

---

🚀 Quickstart

Create a workflow by wrapping functions with @task and @workflow. Awaited tasks automatically map dependencies based on their execution order!

import asyncio
from pyrelay import task, workflow, WorkflowRun
from pyrelay.dashboard import ConsoleDashboard

@task(retries=2, retry_delay=0.5)
async def fetch_user_data(user_id: int):
    await asyncio.sleep(0.5)
    return {"id": user_id, "name": "Alice", "status": "active"}

@task()
async def process_account(user: dict):
    await asyncio.sleep(0.3)
    return f"Account processed for {user['name']}"

@workflow(name="Account Pipeline")
async def account_flow(user_id: int):
    user = await fetch_user_data(user_id)
    result = await process_account(user)
    return result

if __name__ == "__main__":
    run = WorkflowRun(account_flow)
    
    # Attach the live CLI dashboard
    ConsoleDashboard(run)
    
    # Run the workflow blocking/sync
    run.run_sync(user_id=101)

---

🌿 Dynamic Branching (Option B)

In pyrelay, branching is simply Python. The engine tracks exactly which path was taken and marks skipped branches appropriately in reports.

@task()
async def process_premium_user(user: dict):
    return "VIP treatment applied"

@task()
async def process_standard_user(user: dict):
    return "Standard access configured"

@workflow(name="User Routing")
async def user_routing_flow(user: dict):
    # Branching happens at runtime based on real data values
    if user["is_premium"]:
        result = await process_premium_user(user)
    else:
        result = await process_standard_user(user)
    return result

---

🎨 Visualization Telemetry

Generating an interactive, visual representation of your executed workflow is a single method call:

# Save interactive HTML dashboard
run.visualize("workflow_run.html")

Open workflow_run.html in any browser to get a premium dark-mode interface where you can:

1. Pan and zoom around the execution DAG.
2. Review node colors indicating statuses: Green (Success), Red (Failed), Blue (Running), Grey (Pending).
3. Click any node to slide open a telemetry inspect panel containing:
   • Inputs and outputs
   • Time elapsed, start/end timestamps
   • Retry counts and exceptions

---

⚙️ Configuration Properties

The @task decorator accepts the following metadata options:

Property	Type	Default	Description
name	str	Function Name	Identifier for the task (used in dashboard and graphs).
retries	int	0	Number of attempts to retry if the function raises an exception.
retry_delay	float	1.0	Initial delay (seconds) before attempting a retry.
backoff_factor	float	2.0	Exponential multiplier for consecutive retries (delay * backoff_factor^attempt).
timeout	float	None	Max seconds before cancelling execution and marking the task as failed.

---

🧪 Running Tests

To run the unit and integration test suite, install development dependencies and run pytest:

pip install -e ".[dev]"
pytest tests/

---

📄 License

Distributed under the MIT License. See LICENSE for details.