1# Pipelines API Reference
2 
3Code templates and verified behavior. For the full SQL function set and HTTP status semantics, pull `https://developers.cloudflare.com/pipelines/sql-reference/` and the streams docs.
4 
5## Worker Binding Interface
6 
7```typescript
8// from cloudflare:pipelines / @cloudflare/workers-types
9interface Pipeline<T = any> { send(records: T[]): Promise<void>; }
10 
11interface Env { MY_STREAM: Pipeline; }
12 
13export default {
14  async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
15    await env.MY_STREAM.send([{ event_id: crypto.randomUUID(), amount: 29.99 }]);
16    return new Response("OK");
17  }
18} satisfies ExportedHandler<Env>;
19```
20 
21- `send()` takes an **array**, returns `Promise<void>` (no confirmation payload).
22- Throws on network errors — wrap in try/catch or use `ctx.waitUntil()` for fire-and-forget.
23- Validation errors are **not** thrown here (deferred during processing — see [gotchas.md](gotchas.md)).
24- Payload/rate limits apply — check `https://developers.cloudflare.com/pipelines/platform/limits/` before sizing batches.
25 
26## HTTP Ingest
27 
28```
29https://{stream-id}.ingest.cloudflare.com
30```
31 
32Get `{stream-id}` from `npx wrangler pipelines streams list`.
33 
34```bash
35# Batch (preferred)
36curl -X POST https://{stream-id}.ingest.cloudflare.com \
37  -H "Content-Type: application/json" \
38  -d '[{"event_id":"evt-1","amount":29.99},{"event_id":"evt-2","amount":14.99}]'
39 
40# Single event — auto-wrapped in an array
41curl -X POST https://{stream-id}.ingest.cloudflare.com \
42  -H "Content-Type: application/json" -d '{"event_id":"evt-3","amount":9.99}'
43```
44 
45If stream auth is enabled, add `-H "Authorization: Bearer $TOKEN"` (token needs **Workers Pipelines Send**). Standard HTTP status codes apply (400 invalid, 401 auth, 413 too large, 429 rate-limited, 5xx retry).
46 
47> **JSON only** — no Avro, Protobuf, or CSV input.
48 
49## REST Management API
50 
51Base: `https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/pipelines/v1`
52 
53```bash
54# List
55curl -s "$BASE_URL/streams"   -H "Authorization: Bearer $API_TOKEN"
56curl -s "$BASE_URL/sinks"     -H "Authorization: Bearer $API_TOKEN"
57curl -s "$BASE_URL/pipelines" -H "Authorization: Bearer $API_TOKEN"
58 
59# Get one (pipeline GET includes status + failure_reason — useful for debugging)
60curl -s "$BASE_URL/pipelines/{pipeline-id}" -H "Authorization: Bearer $API_TOKEN"
61 
62# Delete in reverse order: pipeline → sink → stream
63curl -X DELETE "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"
64curl -X DELETE "$BASE_URL/sinks/{id}"     -H "Authorization: Bearer $API_TOKEN"
65curl -X DELETE "$BASE_URL/streams/{id}"   -H "Authorization: Bearer $API_TOKEN"
66```
67 
68> `wrangler pipelines delete` defaults to "no" non-interactively — use the REST API for automated cleanup. Deleting a stream removes buffered events and dependent pipelines.
69 
70### Pipeline Lifecycle States
71 
72| Status | Meaning |
73|--------|---------|
74| `running` | Active, processing events |
75| `initializing` | Starting up (minutes after creation or recovery) |
76| `failed` | Stopped on error — check `failure_reason` (expired token, deleted bucket, disabled catalog) |
77 
78> A `GET` on a sink shows `schema.fields: []` — expected. The sink inherits schema from the stream via the pipeline SQL.
79 
80## Pipeline SQL (Transforms)
81 
82Row-level only — no GROUP BY/aggregation. CTEs (`WITH`) and `UNNEST` are supported. Full function list: `https://developers.cloudflare.com/pipelines/sql-reference/`.
83 
84```sql
85-- Passthrough / filter / enrich
86INSERT INTO my_sink SELECT * FROM my_stream;
87INSERT INTO my_sink SELECT * FROM my_stream WHERE amount > 10;
88INSERT INTO my_sink
89SELECT event_id, UPPER(category) AS category, amount * 1.1 AS amount_with_tax
90FROM my_stream;
91 
92-- CTE
93WITH filtered AS (SELECT event_id, amount FROM my_stream WHERE amount > 50)
94INSERT INTO my_sink SELECT * FROM filtered;
95 
96-- UNNEST arrays (one per SELECT)
97SELECT UNNEST(tags) AS tag FROM my_stream;
98```
99 
100Supported categories: string, regex, hashing (`sha256`), JSON extraction, timestamp conversion, conditional (`CASE`), `CAST`, `COALESCE`, math/comparison operators.
101 
102## Verifying End-to-End Data Flow
103 
104```bash
105# 1. Pipeline running (not initializing/failed)?
106curl -s "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"
107 
108# 2. Table created yet? (3–7 min on first flush)
109curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET/namespaces/my_ns/tables" \
110  -H "Authorization: Bearer $API_TOKEN"
111 
112# 3. Data present? (R2 SQL)
113curl -s -X POST \
114  "https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/r2-sql/query/$BUCKET" \
115  -H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \
116  -d '{"query": "SELECT COUNT(*) AS total FROM my_ns.my_table"}'
117```
118 
119> Expect **3–7 minutes** from first send to first queryable data. Subsequent flushes are much faster.
120 
121## See Also
122 
123- [configuration.md](configuration.md) — creating resources · [patterns.md](patterns.md) — producers, Logpush, observability
124- [r2-sql/api.md](../r2-sql/api.md) — querying results
125