Metadata-Version: 2.4
Name: pipely-core
Version: 0.1.3
Summary: Embedded async pipeline engine for FastAPI
Project-URL: Homepage, https://github.com/pipely/pipely
Project-URL: Repository, https://github.com/pipely/pipely
Project-URL: Bug Tracker, https://github.com/pipely/pipely/issues
Author: Pipely Contributors
License: MIT
License-File: LICENSE
Keywords: async,background-tasks,fastapi,pipeline,workflow
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: FastAPI
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.12
Requires-Dist: anyio>=4.0
Requires-Dist: asyncpg>=0.29
Requires-Dist: fastapi>=0.110
Requires-Dist: pydantic>=2.6
Requires-Dist: sqlalchemy[asyncio]>=2.0
Requires-Dist: uvicorn>=0.27
Provides-Extra: test
Requires-Dist: aiosqlite>=0.19; extra == 'test'
Requires-Dist: anyio[trio]>=4.0; extra == 'test'
Requires-Dist: httpx>=0.27; extra == 'test'
Requires-Dist: pytest-anyio>=0.0.0; extra == 'test'
Requires-Dist: pytest>=8.0; extra == 'test'
Description-Content-Type: text/markdown

# pipely-core

[![PyPI](https://img.shields.io/pypi/v/pipely-core)](https://pypi.org/project/pipely-core/)
[![Python](https://img.shields.io/pypi/pyversions/pipely-core)](https://pypi.org/project/pipely-core/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)

Embedded async pipeline engine for FastAPI. For long in-process workflows — OCR, document processing, LLM agents, report generation — where Airflow or Prefect would be too heavy.

- Linear async pipelines with retries and timeouts
- Postgres persistence via SQLAlchemy 2 async (SQLite for tests)
- Automatic table creation — no migrations needed to start
- Resume, retry failed, restart from step, cancel
- Built-in REST API and developer UI

## Install

```bash
pip install pipely-core
```

## Quickstart

```python
from fastapi import FastAPI
from pipely import Pipely, PipelineContext, step

app = FastAPI()
pipely = Pipely(database_url="postgresql+asyncpg://user:password@host:5432/db")
pipely.mount(app)


@pipely.pipeline("meeting_protocol")
class MeetingProtocolPipeline:
    @step(retries=3)
    async def transcribe(self, ctx: PipelineContext) -> None:
        transcript = await transcribe_audio(ctx.input["audio_url"])
        ctx.set("transcript", transcript)

    @step(retries=2)
    async def clean_transcript(self, ctx: PipelineContext) -> None:
        clean = await clean_text(ctx.get("transcript"))
        ctx.set("clean_transcript", clean)

    @step(retries=2)
    async def extract_tasks(self, ctx: PipelineContext) -> None:
        tasks = await extract_tasks(ctx.get("clean_transcript"))
        ctx.set("tasks", tasks)
```

## Running pipelines

**Blocking** — waits until the run finishes:
```python
run = await pipely.start("meeting_protocol", input={"audio_url": "s3://bucket/audio.mp3"})
print(run.status)   # "completed"
print(run.state)    # {"transcript": ..., "tasks": [...]}
```

**Background** — returns immediately, runs in the app's task group:
```python
run = await pipely.submit("meeting_protocol", input={"audio_url": "..."})
print(run.status)   # "pending"
```

## Controlling runs

```python
await pipely.retry_failed(run.id)                        # retry from first failed step
await pipely.resume(run.id)                              # resume from first unfinished step
await pipely.restart_from_step(run.id, "extract_tasks")  # restart from a specific step
await pipely.cancel(run.id)                              # cancel immediately
```

## PipelineContext

Each step receives a `ctx` object:

| Member | Description |
|---|---|
| `ctx.input` | Original input dict passed to `start()` |
| `ctx.state` | Mutable dict shared across all steps, persisted to DB |
| `ctx.get(key)` | Read from state |
| `ctx.set(key, value)` | Write to state (auto-flushed after step) |
| `await ctx.flush()` | Persist state mid-step without waiting for step to finish |
| `await ctx.log(msg, level, payload)` | Write a structured log entry |

```python
@step(retries=2, timeout=30.0)
async def process(self, ctx: PipelineContext) -> None:
    await ctx.log("starting", payload={"input": ctx.input})
    result = await do_work(ctx.get("data"))
    ctx.set("result", result)
    await ctx.flush()  # persist now, in case the next part fails
    await do_more_work()
```

## Composing with an existing lifespan

```python
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    await startup()
    async with pipely.lifespan(app):
        yield
    await shutdown()

app = FastAPI(lifespan=lifespan)
pipely.mount(app)
```

## Using an existing SQLAlchemy engine

```python
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(DATABASE_URL)
pipely = Pipely(engine=engine)
# or pass your own sessionmaker:
pipely = Pipely(sessionmaker=async_sessionmaker(engine, expire_on_commit=False))
```

## REST API & UI

`pipely.mount(app)` adds these routes (default prefix `/_pipely`):

| Method | Path | Description |
|---|---|---|
| `GET` | `/_pipely` | Developer UI |
| `GET` | `/_pipely/api/runs` | List runs |
| `GET` | `/_pipely/api/runs/{run_id}` | Get run with steps and logs |
| `POST` | `/_pipely/api/runs` | Start a run |
| `POST` | `/_pipely/api/runs/{run_id}/retry` | Retry from failed step |
| `POST` | `/_pipely/api/runs/{run_id}/resume` | Resume from first unfinished step |
| `POST` | `/_pipely/api/runs/{run_id}/restart-from-step` | Restart from a named step |
| `POST` | `/_pipely/api/runs/{run_id}/cancel` | Cancel a run |

Change the prefix:
```python
pipely = Pipely(database_url=..., ui_path="/admin/pipelines")
```

## Development

```bash
git clone https://github.com/pipely/pipely
cd pipely
pip install -e ".[test]"
pytest
```
