pipely-core 0.1.0

Embedded async pipeline engine for FastAPI

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

pip install -e ".[test]"

For production use, provide a Postgres URL:

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

Usage

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:

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.