codeupipe¶

Python pipeline framework — composable Payload → Filter → Pipeline pattern with streaming. Zero external dependencies.

Get Started in 5 Minutes

Install codeupipe and build your first pipeline in minutes.

Quick Start
Concepts & API

Full reference for every type — Payload, Filter, Pipeline, Valve, Tap, and more.

Concepts
Best Practices

Project structure, naming conventions, testing strategy, CLI workflow.

Best Practices
Deploy

Docker, Render (free), Vercel, Netlify — from cup.toml to running app.

Deploy Guide
Connect Platform

Device mesh — mobile → browser extension → desktop compute. CUP pipelines at every hop.

Platform
Roadmap

13 completed rings and open directions — from pipeline framework to connected device mesh.

Roadmap
cup/core Design System

20 Web Components + layered CSS. Zero dependencies, browser-native.

Documentation · Playground

Core Concepts at a Glance¶

Concept	Role
Payload	Immutable data container flowing through the pipeline
MutablePayload	Mutable sibling for performance-critical bulk edits
Filter	Processing unit — sync or async `.call(payload) → Payload`
StreamFilter	Streaming — async `.stream(payload)` yields 0..N chunks
Pipeline	Orchestrator — `.run()` for batch, `.stream()` for streaming
Valve	Conditional flow control — gates a Filter with a predicate
Tap	Non-modifying observation — inspect without changing
State	Execution metadata — tracks what ran, skipped, errors
Hook	Lifecycle — `before()` / `after()` / `on_error()`
RetryFilter	Resilience — retries a Filter up to N times

The Device Mesh¶

codeupipe connects every device into a single pipeline mesh.

Mobile  ──▶  Platform SPA  ──▶  Browser Extension  ──▶  Desktop / Servers
  (phone)     (static page)       (MV3 bridge)          (DB, GPU, files)

A phone triggers a static web page. The page talks to a browser extension. The extension relays to a native host on the physical machine — with full access to databases, GPU compute, and local files. Every hop is a CUP pipeline. The same mesh works in reverse. Learn more

Minimal Example¶

import asyncio
from codeupipe import Payload, Pipeline

class CleanInput:
    def call(self, payload):
        return payload.insert("text", payload.get("text", "").strip())

class Validate:
    def call(self, payload):
        if not payload.get("text"):
            raise ValueError("Empty input")
        return payload

pipeline = Pipeline()
pipeline.add_filter(CleanInput(), name="clean")
pipeline.add_filter(Validate(), name="validate")

result = asyncio.run(pipeline.run(Payload({"text": "  hello  "})))
print(result.get("text"))  # "hello"

Install¶

pip install codeupipe

Zero Dependencies

codeupipe uses only the Python standard library. No external packages required. Python 3.9+.