Metadata-Version: 2.5
Name: promiseflow
Version: 0.2.0
Summary: A Python implementation of promise-based parallel processing for coordinating asynchronous and concurrent workloads.
Project-URL: Homepage, https://github.com/srathbun/promiseflow
Project-URL: Repository, https://github.com/srathbun/promiseflow
Project-URL: Issues, https://github.com/srathbun/promiseflow/issues
Author: Spencer Rathbun
License: MIT
License-File: LICENSE
Keywords: asyncio,concurrency,distributed-systems,promises,singleflight
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.10
Provides-Extra: dev
Requires-Dist: fakeredis>=2.26.0; extra == 'dev'
Requires-Dist: lupa>=2.2; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: redis>=5.0.0; extra == 'dev'
Requires-Dist: ruff>=0.5.0; extra == 'dev'
Provides-Extra: redis
Requires-Dist: redis>=5.0.0; extra == 'redis'
Description-Content-Type: text/markdown

# promiseflow

PromiseFlow is a Python implementation of the promise-based parallel processing model described in ACM Queue *Parallel Processing with Promises*. It provides composable promises for coordinating asynchronous and parallel computation, allowing complex workflows to be expressed as chains of dependent operations where duplicate work is automatically eliminated.

- one owner performs a keyed unit of work
- followers join the same future and receive the result without repeating work
- stale workers are timed out by heartbeat sweeps
- retries happen through a simple policy
- composable chains let multiple callers share intermediate results

## Install

```bash
pip install promiseflow
```

## Quickstart

### Single unit of work

Use `Coordinator.get_or_run` directly when you have a single keyed operation
that many callers might request concurrently:

```python
import asyncio
from promiseflow import Coordinator, RetryPolicy


async def main() -> None:
    async with Coordinator(stale_after=3.0, sweep_interval=0.5) as coordinator:
        async def expensive() -> str:
            await asyncio.sleep(0.2)
            return "done"

        result = await coordinator.get_or_run(
            "job:42",
            expensive,
            timeout=5.0,
            retry=RetryPolicy(max_attempts=3),
            heartbeat_interval=0.25,
        )
        print(result)


asyncio.run(main())
```

If another caller requests `"job:42"` while the first is still running, it
hooks onto the same future and waits — no duplicate work. After that run
finishes, a later request for the same key starts fresh work (ephemeral
retention; see [Semantics](#semantics)).

### Composable chains

The real power shows up when work can be broken into named segments that are
shared across callers.  `Chain` hashes the initial input together with the
ordered sequence of step names to build a deduplication key for each segment.
Two concurrent chains that share a prefix automatically share intermediate
results. Step names are the identity — reuse a name only when the work is
meant to be shared.

Consider a data pipeline where several users issue queries against a database.
Each query is a pipeline of operations — scan, sort, group, limit — and many
queries share a common prefix:

```python
import asyncio
from promiseflow import Coordinator, Chain


async def scan(params):
    """Simulate an expensive database scan."""
    await asyncio.sleep(1.0)
    return [{"x": 1}, {"x": 2}, {"x": 3}]


async def sort_rows(rows):
    """Sort results by x."""
    return sorted(rows, key=lambda r: r["x"])


async def group_rows(rows):
    """Group/aggregate the sorted results."""
    return {"count": len(rows), "sum": sum(r["x"] for r in rows)}


async def limit_rows(rows):
    """Return only the first two rows."""
    return rows[:2]


async def main() -> None:
    async with Coordinator(stale_after=10.0, sweep_interval=1.0) as coordinator:

        # User A: scan → sort → group
        async def user_a():
            return await (
                Chain(coordinator)
                .add("scan", scan)
                .add("sort", sort_rows)
                .add("group", group_rows)
                .run()
            )

        # User B: scan → sort → limit
        async def user_b():
            return await (
                Chain(coordinator)
                .add("scan", scan)
                .add("sort", sort_rows)
                .add("limit", limit_rows)
                .run()
            )

        # Both users run concurrently.  The scan and sort steps execute
        # exactly once even though two users requested them.
        result_a, result_b = await asyncio.gather(user_a(), user_b())

        print("User A (group):", result_a)
        # -> {'count': 3, 'sum': 6}

        print("User B (limit):", result_b)
        # -> [{'x': 1}, {'x': 2}]


asyncio.run(main())
```

In this example:

- **`scan`** runs once.  Both users share the result.
- **`sort`** runs once.  Both users share the sorted output.
- **`group`** and **`limit`** each run once — they diverge at this point, so
  each gets its own deduplication key.

This is exactly how the original implementation worked for MongoDB aggregation
pipelines: each operation in the aggregation pipeline was a named step, and
the coordinator ensured that concurrent queries with shared prefixes didn't
re-run the same expensive database scans.

### Writing your own step functions

A step function is any `async` callable that takes one argument (the output
of the previous step) and returns a value:

```python
async def my_step(previous_result):
    # Do work with previous_result ...
    return transformed_result
```

Steps are composed by name so the coordinator can tell them apart. Identical
names with the same initial input share work under concurrency, even if the
callables differ — choose unique names when steps are different:

```python
chain = (
    Chain(coordinator)
    .add("fetch",     fetch_from_db)
    .add("transform", apply_business_rules)
    .add("enrich",    call_external_api)
)
result = await chain.run(initial={"query": "..."})
```

If you need retries within a chain (for example, a step that calls a flaky
external service), pass a `RetryPolicy`:

```python
from promiseflow import RetryPolicy

result = await chain.run(
    initial={"query": "..."},
    retry=RetryPolicy(max_attempts=3, base_delay=0.1),
)
```

By default, chain steps use `max_attempts=1` so errors surface immediately
rather than being silently retried.

## Semantics

### Ephemeral retention

PromiseFlow is a **single-flight** coordinator by default, not a result cache:

- Concurrent callers for the same key share one in-flight execution.
- When that execution completes (success or failure), the in-flight entry is
  dropped.
- A later call with the same key runs the work again.

Retention is an explicit policy (`promiseflow.Ephemeral` is the default). Longer
lived policies (`ttl`, manual invalidate, stale-while-revalidate) are tracked in
[`docs/plan.md`](docs/plan.md).

### Chain segment keys

Segment keys are `hash(initial, ordered step names)`. Callable identity is not
part of the key. Treat step names as the public contract for what may be shared.

## Distributed coordination (0.2)

`RedisCoordinator` gives the same single-flight semantics across processes and
servers, backed by Redis:

```python
import asyncio
import redis.asyncio as redis

from promiseflow import RedisCoordinator, RedisBackend, RetryPolicy


async def main() -> None:
    client = redis.from_url("redis://localhost:6379")
    backend = RedisBackend(client, namespace="my-app")

    async with RedisCoordinator(backend) as coordinator:
        async def expensive() -> dict:
            await asyncio.sleep(0.5)
            return {"computed": True}

        result = await coordinator.get_or_run(
            "job:42",
            expensive,
            retry=RetryPolicy(max_attempts=3),
        )
        print(result)


asyncio.run(main())
```

Cliff notes:

- **One owner per key** acquires a Redis lock (`SET NX`); followers in other
  processes join the same generation.
- **Lock TTL + heartbeat** reclaim dead owners (no polling; a stale lock simply
  expires).
- **Temporary payload key** carries the result to followers; pub/sub `BUILT` /
  `FAILED` messages are the wake signal. Payloads are generation-scoped and
  evicted by `Ephemeral.payload_ttl`.
- Every call returns the owner's result by value, so results must be serializable
  by the backend codec (pickle by default; pluggable).
- `Chain` works unchanged with `RedisCoordinator`.
- `LockBackend` / `PayloadStore` / `MessageBus` protocols make the backend
  replaceable; `RedisBackend` is the shipped implementation.

## Status

Version `0.2.0` adds distributed Redis coordination (`RedisCoordinator`,
`RedisBackend`, pluggable protocols) and a first-class `RetentionPolicy`.
Remaining retention policies (`ttl`, manual, stale-while-revalidate) are tracked
in [`docs/plan.md`](docs/plan.md) and cut in [`docs/roadmap.md`](docs/roadmap.md)
(`0.3.0`).
