Metadata-Version: 2.5
Name: promiseflow
Version: 0.1.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: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest>=8.0.0; extra == 'dev'
Requires-Dist: ruff>=0.5.0; extra == 'dev'
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 (0.1)

PromiseFlow 0.1 is a **single-flight** coordinator, 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.

Longer-lived retention (`ttl`, manual invalidate, stale-while-revalidate) is
planned as an explicit policy; see [`docs/plan.md`](docs/plan.md).

### Chain segment keys (0.1)

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.

## Status

Version `0.1.0` is in-process only with ephemeral retention. Distributed Redis
coordination and pluggable result retention are tracked in
[`docs/plan.md`](docs/plan.md) and cut in [`docs/roadmap.md`](docs/roadmap.md)
(`0.2.0` / `0.3.0`).
