flowlet 0.1.3
A lightweight Python workflow engine with DAG nodes, async execution, and conditional branches.
Requires Python: >=3.8

# flowlet

A lightweight Python workflow engine with DAG nodes, async execution, conditional branches, and optional dependencies.

## Features

- Function-based node definition with dependency resolution
- Concurrent execution within each DAG level (asyncio)
- Conditional branches via `when`
- Optional dependencies via `optional(...)`
- Runtime inputs injected from `WORKFLOW_PARAM` (JSON)
- Execution context: trace_id, run_id, timings, logs, outputs
- Exportable workflow graph via `workflow_compile_graph`
- Optional hook events via `FLOWLET_HOOK_URL`

## Installation

```bash
pip install flowlet
```

## Quickstart

```python
import asyncio
from flowlet import Input, node, optional, workflow_compile, workflow_run

class Inputs:
    a = Input(int, desc="param a")
    b = Input(int, desc="param b")

@node(inputs={"x": Inputs.a}, outputs={"result": "x"})
async def step1(x):
    await asyncio.sleep(0.1)
    return x + 1

@node(inputs={"y": Inputs.b}, outputs={"result": "y"})
def step2(y):
    return y * 2

@node(inputs={"x": step1.result, "y": step2.result}, outputs={"route": "branch"})
def route(x, y):
    return "A" if x + y >= 0 else "B"

@node(
    inputs={"route": route.route, "x": step1.result},
    outputs={"result": "A"},
    when=lambda route, **_: route == "A",
)
def step_a(route, x):
    return x * 10

@node(
    inputs={"route": route.route, "y": step2.result},
    outputs={"result": "B"},
    when=lambda route, **_: route == "B",
)
def step_b(route, y):
    return y * -10

@node(
    inputs={"a": optional(step_a.result), "b": optional(step_b.result)},
    outputs={"result": "merge"},
)
def merge(a=None, b=None):
    return a if a is not None else b

compiled = workflow_compile(Inputs)
ctx, output = workflow_run(compiled)
print(output)
```

Provide runtime inputs via environment variable:

```bash
export WORKFLOW_PARAM='{"a": 1, "b": 2}'
python your_app.py
```

Or pass params directly:

```python
ctx, output = workflow_run(compiled, params={"a": 1, "b": 2})
```

## Concepts

### Inputs

Declare inputs with `Input(type, desc)` and provide values through `WORKFLOW_PARAM`. Types are cast at runtime.

### Node

Use `@node(inputs=..., outputs=..., when=..., name=...)` to wrap a function:

- `inputs`: mapping of parameter name to `Input` or upstream output
- `outputs`: output names and descriptions
- `when`: callable that returns True/False to control execution
- `name`: optional friendly node name (default is function name, must be unique)

### Optional dependencies

Use `optional(step.result)` for dependencies that may be missing; they are injected as `None`.

### Execution

`workflow_compile(Inputs)` builds a compiled graph. `workflow_run(compiled)` runs the workflow and returns:

- `ctx` with trace_id, run_id, timings, logs, outputs, skipped
- `output` is the last node's output

### Graph export

`workflow_compile_graph(Inputs)` returns a serializable DAG:

- `inputs`: input definitions
- `nodes`: node metadata including docstring and source
- `edges`: dependency edges
- `levels`: parallel execution levels

### Hook events (optional)

Set `FLOWLET_HOOK_URL` to enable non-blocking hook events (HTTP POST, JSON). If unset, no hook logic runs.

Environment variables:

- `FLOWLET_HOOK_URL`: destination URL for hook events
- `FLOWLET_HOOK_TIMEOUT`: per-request timeout in seconds (default: 1.0)
- `FLOWLET_HOOK_MAX_QUEUE`: max in-memory queue for hook events (default: 1000)
- `FLOWLET_WORKFLOW_ID`: external workflow id to attach to all events
- `FLOWLET_HOOK_FLUSH_TIMEOUT`: best-effort flush time on exit / run end (seconds, default: 0.5)

Events:

- `workflow.graph`: emitted during compile, includes graph (with node source)
- `workflow.run_start`: emitted at start of execution
- `workflow.level_done`: emitted after each DAG level completes
- `workflow.run_end`: emitted at end of execution (status + output)

Common fields on all events:

- `schema_version`: currently `1`
- `workflow_hash`: hash of the compiled graph
- `workflow_id`: from `FLOWLET_WORKFLOW_ID` if set
- `workflow_path`: absolute path to the workflow script (when available)
- `timestamp`: seconds since epoch

Minimal example payload:

```json
{
  "event": "workflow.level_done",
  "schema_version": 1,
  "workflow_hash": "…",
  "workflow_id": "wf_123",
  "workflow_path": "/abs/path/to/workflow.py",
  "trace_id": "…",
  "run_id": "…",
  "level_index": 0,
  "timestamp": 1730000000.0,
  "nodes": [
    {
      "name": "step1",
      "status": "ok",
      "elapsed": 0.0123,
      "outputs": {"result": 42},
      "logs": []
    }
  ]
}
```

## Debug UI (dev only)

Install with dev extras:

```bash
pip install flowlet[dev]
```

In your workflow script, start the dev server instead of calling `workflow_compile` / `workflow_run`:

```python
from flowlet.dev import workflow_server

if __name__ == "__main__":
    workflow_server(Inputs)
```

Then open:

```
http://127.0.0.1:8777
```

Use the Runner panel to compile the graph and run with JSON params.

## Notes

- `WORKFLOW_PARAM` must be valid JSON.
- For a single output, return a scalar. For multiple outputs, return a dict or tuple/list.

## License

MIT