pipely-core 0.1.4

Embedded async pipeline engine for FastAPI

pipely-core

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

pip install pipely-core

Quickstart

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:

run = await pipely.start("meeting_protocol", input={"audio_url": "s3://bucket/audio.mp3"})
print(run.status)   # "completed"
print(run.state)    # {"transcript": ..., "tasks": [...]}

With timeout:

run = await pipely.start("long_task", input={...}, timeout=300.0)  # 5 minutes max

Background — returns immediately, runs in the app's task group:

run = await pipely.submit("meeting_protocol", input={"audio_url": "..."})
print(run.status)   # "pending"

Controlling runs

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
await pipely.retry_all_failed("meeting_protocol")       # retry all failed runs of a pipeline

Production features

Graceful shutdown — waits for active runs before stopping:

# Automatically handled in lifespan, or call manually:
await pipely.executor.shutdown(timeout=30.0)

Auto-cleanup old runs:

pipely = Pipely(database_url=..., retention_days=30)  # delete runs older than 30 days

Health check:

curl http://localhost:8000/_pipely/api/health
# {"status": "healthy", "database": "ok", "stuck_runs": 0}

Dead letter queue — failed runs are automatically saved for review:

curl http://localhost:8000/_pipely/api/failed-runs

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

@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

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

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:

pipely = Pipely(database_url=..., ui_path="/admin/pipelines")

Development

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