API

// from cloudflare:pipelines / @cloudflare/workers-types
interface Pipeline<T = any> { send(records: T[]): Promise<void>; }

interface Env { MY_STREAM: Pipeline; }

export default {
  async fetch(req: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
    await env.MY_STREAM.send([{ event_id: crypto.randomUUID(), amount: 29.99 }]);
    return new Response("OK");
  }
} satisfies ExportedHandler<Env>;

https://{stream-id}.ingest.cloudflare.com

# Batch (preferred)
curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" \
  -d '[{"event_id":"evt-1","amount":29.99},{"event_id":"evt-2","amount":14.99}]'

# Single event — auto-wrapped in an array
curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" -d '{"event_id":"evt-3","amount":9.99}'

# List
curl -s "$BASE_URL/streams"   -H "Authorization: Bearer $API_TOKEN"
curl -s "$BASE_URL/sinks"     -H "Authorization: Bearer $API_TOKEN"
curl -s "$BASE_URL/pipelines" -H "Authorization: Bearer $API_TOKEN"

# Get one (pipeline GET includes status + failure_reason — useful for debugging)
curl -s "$BASE_URL/pipelines/{pipeline-id}" -H "Authorization: Bearer $API_TOKEN"

# Delete in reverse order: pipeline → sink → stream
curl -X DELETE "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"
curl -X DELETE "$BASE_URL/sinks/{id}"     -H "Authorization: Bearer $API_TOKEN"
curl -X DELETE "$BASE_URL/streams/{id}"   -H "Authorization: Bearer $API_TOKEN"

-- Passthrough / filter / enrich
INSERT INTO my_sink SELECT * FROM my_stream;
INSERT INTO my_sink SELECT * FROM my_stream WHERE amount > 10;
INSERT INTO my_sink
SELECT event_id, UPPER(category) AS category, amount * 1.1 AS amount_with_tax
FROM my_stream;

-- CTE
WITH filtered AS (SELECT event_id, amount FROM my_stream WHERE amount > 50)
INSERT INTO my_sink SELECT * FROM filtered;

-- UNNEST arrays (one per SELECT)
SELECT UNNEST(tags) AS tag FROM my_stream;

# 1. Pipeline running (not initializing/failed)?
curl -s "$BASE_URL/pipelines/{id}" -H "Authorization: Bearer $API_TOKEN"

# 2. Table created yet? (3–7 min on first flush)
curl -s "https://api.cloudflare.com/client/v4/accounts/$ACCOUNT_ID/r2-catalog/$BUCKET/namespaces/my_ns/tables" \
  -H "Authorization: Bearer $API_TOKEN"

# 3. Data present? (Basin SQL)
curl -s -X POST \
  "https://api.sql.cloudflarestorage.com/api/v1/accounts/$ACCOUNT_ID/basin-sql/query/$BUCKET" \
  -H "Authorization: Bearer $API_TOKEN" -H "Content-Type: application/json" \
  -d '{"query": "SELECT COUNT(*) AS total FROM my_ns.my_table"}'