Metadata-Version: 2.4
Name: pipely-core
Version: 0.1.0
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

Pipely is a small embedded async pipeline engine for FastAPI services. It is for long in-process workflows such as OCR, document processing, meeting protocol generation, indexing, LLM agents, and report generation where Airflow or Prefect would be too heavy.

MVP scope:

- linear async pipelines;
- Postgres persistence through SQLAlchemy 2 async;
- automatic table creation;
- in-process execution;
- retries, resume, retry failed, restart from step, cancel;
- FastAPI REST API and a small built-in HTML UI.

## Install

```bash
pip install -e ".[test]"
```

For production use, provide a Postgres URL:

```text
postgresql+asyncpg://user:password@host:5432/database
```

## Usage

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

app = FastAPI()

pipely = Pipely(database_url=settings.DATABASE_URL, ui_path="/_pipely")
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)
```

Start and control runs:

```python
run = await pipely.start("meeting_protocol", input={"audio_url": "s3://bucket/audio.mp3"})

await pipely.retry_failed(run.id)
await pipely.resume(run.id)
await pipely.restart_from_step(run.id, "extract_tasks")
await pipely.cancel(run.id)
```

Mounted endpoints:

- `GET /_pipely`
- `GET /_pipely/api/runs`
- `GET /_pipely/api/runs/{run_id}`
- `POST /_pipely/api/runs`
- `POST /_pipely/api/runs/{run_id}/retry`
- `POST /_pipely/api/runs/{run_id}/resume`
- `POST /_pipely/api/runs/{run_id}/restart-from-step`
- `POST /_pipely/api/runs/{run_id}/cancel`

## Architecture

- `Pipely` registers pipelines and mounts FastAPI routes.
- `PipelineContext` exposes `input`, persistent `state`, `get`, `set`, `flush`, and `log`.
- `Storage` owns SQLAlchemy async persistence and creates tables from metadata.
- `InProcessExecutor` executes steps sequentially, skips successful steps on resume, records attempts/errors/tracebacks, and marks stale running steps failed before resume.
- `ui.py` serves a dependency-free HTML/JS developer UI.

## Next Steps

- background execution mode for REST starts;
- real migration integration with Alembic;
- graph transitions after the linear MVP stabilizes;
- cron/queue/distributed workers;
- richer metrics export.
