Metadata-Version: 2.5
Name: fluss
Version: 6.0.0
Summary: the next generation of Fluss, a Python library for workflow design
Author-email: jhnnsrs <jhnnsrs@gmail.com>
License-Expression: GPL-3.0-or-later
License-File: LICENSE
Requires-Python: <4.0,>=3.11
Requires-Dist: arkitekt-spec>=3.4
Requires-Dist: fakts>=5.4
Requires-Dist: rath>=4.1.1
Requires-Dist: websockets>=15.0.1
Provides-Extra: engine
Requires-Dist: arkitekt-runtime>=4.3; extra == 'engine'
Requires-Dist: rekuest>=8.0.1; extra == 'engine'
Provides-Extra: python
Requires-Dist: arkitekt-runtime>=4.3; extra == 'python'
Requires-Dist: fluss-monty>=1; extra == 'python'
Requires-Dist: rekuest>=8.0.1; extra == 'python'
Description-Content-Type: text/markdown

# fluss

[![PyPI version](https://badge.fury.io/py/fluss.svg)](https://pypi.org/project/fluss/)
[![Maintenance](https://img.shields.io/badge/Maintained%3F-yes-green.svg)](https://pypi.org/project/fluss/)
![Maintainer](https://img.shields.io/badge/maintainer-jhnnsrs-blue)
[![PyPI pyversions](https://img.shields.io/pypi/pyversions/fluss.svg)](https://pypi.python.org/pypi/fluss/)
[![PyPI download month](https://img.shields.io/pypi/dm/fluss.svg)](https://pypi.python.org/pypi/fluss/)

The python client for fluss, the [Arkitekt](https://arkitekt.live) workflow service: designing,
storing and running workflows (flows) that wire the actions of your apps together.

## Installation

```bash
pip install fluss
```

The flow *executor* is an optional extra, because it needs rekuest while the plain client does not:

```bash
pip install "fluss[engine]"   # run flows (fluss.engine)
pip install "fluss[python]"   # also run flows written as Python source
```

With arkitekt, `pip install "arkitekt[rekuest,fluss]"` brings it in.

## Design

A **flow** is a graph: argument nodes feed reactive and action nodes, whose outputs reach return
nodes. Flows live in **workspaces**, and every execution of a flow is a **run**. fluss is the client
for that graph — it queries and mutates flows, workspaces and runs on the fluss server — and
`fluss.engine` is what executes one.

## Usage

Every fluss operation is a method of the `Fluss` client, in a blocking and an `a`-prefixed async
flavour (`fluss.get_flow(id)`, `await fluss.aget_flow(id)`).

### In an arkitekt app

Add the service to your app and ask for `fluss: Fluss`; the client is injected by annotation. Flows
and runs travel between actions by id (`@fluss/flow`, `@fluss/run`, `@fluss/pythonflow`,
`@fluss/pythonrun`), so an action can take and return them directly:

```python
from arkitekt import App, run
from fluss import Fluss, fluss_service
from fluss.api.schema import Flow

app = App("flow-inspector", "0.1.0", services=[fluss_service])


@app.action
def describe_flow(flow: Flow, fluss: Fluss) -> str:
    """Describe Flow"""
    return f"{flow.title} in workspace {flow.workspace.id}"


if __name__ == "__main__":
    run(app)
```

An app built with the fluss service also offers the generic **Run Flow** action (interface
`run_flow`), so it can execute any flow it is handed (**Run Python Flow** too, with the `python`
extra).

### From a script

```python
from arkitekt import easy
from fluss import fluss_service

with easy("my-script", fluss_service) as fluss:
    for flow in fluss.flows(limit=10):
        print(flow.id, flow.title)
```

## The engine

`fluss.engine` executes a flow as a generic rekuest action: given a flow and a dict of arguments it
yields the flow's returns, so a workflow can be called like any other action.

A flow run is registered as a rekuest **workflow**: if the agent running it dies, it is resumed,
and the node calls that already finished return their recorded results instead of running again.
Each call is keyed by its node and how often that node ran, so concurrent branches are found
again whichever order they finish in.

```python
from fluss.engine import run_flow, arun_flow
```

It imports rekuest at module level, so install `fluss[engine]` to use it. Importing plain `fluss`
never requires rekuest.

### Python flows

`fluss.engine.python` (the `python` extra) validates and runs flows written as Python source in a
Monty sandbox: `async def main(...)` calls the actions of its manifest by alias.

- **Run Python Flow** is a workflow too. Each action call is keyed by the action, a digest of its
  arguments and how often that exact call was made before, so a resumed run finds its calls
  again without depending on how its gathered calls interleave. The sandbox's `random` seed is
  fixed per run, so the run makes the same calls again. One case remains: identical calls in
  parallel branches may trade results on resume. The calls downstream of them then run again,
  which repeats an `IRREVERSIBLE` action if one is among them.
- Its actor shares one Monty worker pool across runs. `print` output reaches the task log while
  the run goes on. A cancelled run cancels the calls it has in flight and ends `CANCELLED`.
- Publishing a version makes it runnable, not an action. To call a version like any action,
  deploy it onto an agent that hosts `run_python_flow`:

```python
from fluss.engine.python.client import acreate_python_flow_from_source, adeploy_python_flow, amanifest

actions = await amanifest({"segment": segment_hash}, rekuest)
flow = await acreate_python_flow_from_source(source, actions, fluss, title="Segment", publish=True)
await adeploy_python_flow(flow, rekuest, runner=run_python_flow_implementation_id)
```

## Testing

```bash
uv run pytest -m "not integration"   # no server needed
uv run pytest -m integration          # a real fluss server via dokker
```

See [RELEASING.md](RELEASING.md) for how versions are cut.
