Reading Arrow

The client libraries for TypeScript, Python and Rust do everything on this page for you and return JSON rows. Read on if you want Arrow tables (for Parquet, pandas, Polars, DuckDB) or use another language.

POST /query returns application/vnd.apache.arrow.stream. Any Apache Arrow library can read it. There is one thing to get right: the body is several Arrow IPC streams back to back, one per row type, not one stream.

The response layout

One complete Arrow IPC stream per row type you asked for, always in the order transactions, then logs, then blocks (whatever order you listed them in).
Each stream has its own schema, its record batches, and its own end-of-stream marker.
Every stream’s schema carries the same bl.* metadata. bl.streams lists the streams in order ("transactions,logs") and bl.row_type names the current one.
The first schema arrives as soon as the query is planned, before any rows.

A correct reader:

Opens a stream reader on the body and reads it to its end-of-stream marker.
Reads bl.streams from its schema, then opens the next stream on the same body, until it has read that many.
Checks that the total rows received equal bl.objects.

Check for truncation. If a query fails after the server has started sending, the server can’t send an error body any more: it closes the connection. Most Arrow readers raise an error when a body ends mid-batch, but a body cut exactly between two batches can look like a valid, shorter stream. bl.objects is in the schema, sent before any row, and equals the number of rows the response carries. Compare it with what you received, and treat a mismatch as a failed request. The readers below do this.

Complete readers

Each of these is a small, complete client that returns Arrow tables: it sends the query, decodes every stream, checks for truncation and reports API errors with their code. Copy one into your project.

Python
TypeScript
Rust

Needs pyarrow and requests (pip install pyarrow requests). query() also retries rate limits, busy servers and broken connections; see Errors and retries.

bloklake_arrow.py

"""A small, complete Bloklake client: POST /query, decode every Arrow stream, retry what's retryable."""
import os
import time

import pyarrow as pa
import pyarrow.ipc as ipc
import requests

BASE = os.environ.get("BLOKLAKE_URL", "https://api.bloklake.com/v1/chains/eth")
KEY = os.environ["BLOKLAKE_API_KEY"]
session = requests.Session()
session.headers["Authorization"] = f"Bearer {KEY}"


class QueryError(Exception):
    def __init__(self, status: int, code: str, message: str, retry_after: float | None = None):
        super().__init__(f"{status} {code}: {message}")
        self.status, self.code, self.retry_after = status, code, retry_after


def _query_once(body: dict) -> tuple[dict[str, pa.Table], dict[str, str]]:
    resp = session.post(f"{BASE}/query", json=body, stream=True, timeout=(10, 120))
    if resp.status_code != 200:
        try:
            err = resp.json()["error"]
        except Exception:
            err = {"code": "http_error", "message": resp.text[:200]}
        ra = resp.headers.get("Retry-After")
        raise QueryError(resp.status_code, err["code"], err["message"], float(ra) if ra else None)
    resp.raw.decode_content = True
    tables: dict[str, pa.Table] = {}
    meta: dict[str, str] = {}
    try:
        while True:
            reader = ipc.open_stream(resp.raw)       # one complete Arrow IPC stream
            table = reader.read_all()                # ends at this stream's end-of-stream marker
            meta = {k.decode(): v.decode() for k, v in reader.schema.metadata.items()}
            tables[meta["bl.row_type"]] = table
            if len(tables) == len(meta["bl.streams"].split(",")):
                break
    except (pa.ArrowInvalid, OSError, requests.exceptions.RequestException) as e:
        # the server aborts the connection if a query fails after it started sending
        raise QueryError(0, "truncated", f"response ended early: {e}") from e
    # A body cut exactly between two batches reads as a shorter, valid stream:
    # bl.objects (sent before any row) is the number of rows the response carries.
    received = sum(t.num_rows for t in tables.values())
    if received != int(meta["bl.objects"]):
        raise QueryError(0, "truncated", f"got {received} of {meta['bl.objects']} rows")
    return tables, meta


def query(body: dict, attempts: int = 5) -> tuple[dict[str, pa.Table], dict[str, str]]:
    """Run a query, retrying rate limits, busy servers and broken connections."""
    for attempt in range(attempts):
        try:
            return _query_once(body)
        except QueryError as e:
            retryable = e.status in (429, 503) or e.status >= 500 or e.code == "truncated"
            if not retryable or attempt == attempts - 1:
                raise
            time.sleep(e.retry_after or min(2 ** attempt, 30))
    raise AssertionError("unreachable")

from bloklake_arrow import query

tables, meta = query({
    "fromBlock": 19426000, "toBlock": 19426009,
    "logs": [{"address": ["0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"]}],
    "joinTransactions": True,
    "fields": {"log": {"blockNumber": True, "logIndex": True}, "transaction": {"hash": True}},
})
print({name: t.num_rows for name, t in tables.items()})   # {'transactions': 62, 'logs': 85}
print(meta["bl.next_block"])                              # 19426010

Needs apache-arrow (npm install apache-arrow). Works in Node 18+, Bun and Deno. RecordBatchReader.readAll yields one reader per stream in the body.

bloklake-arrow.ts

import { RecordBatchReader, Table } from "apache-arrow";

const BASE = process.env.BLOKLAKE_URL ?? "https://api.bloklake.com/v1/chains/eth";
const KEY = process.env.BLOKLAKE_API_KEY!;

export class QueryError extends Error {
  constructor(public status: number, public code: string, message: string, public retryAfter?: number) {
    super(`${status} ${code}: ${message}`);
  }
}

/** POST /query: every Arrow stream of the response as a Table, keyed by row type, plus the bl.* metadata. */
export async function query(body: unknown): Promise<{ tables: Record<string, Table>; meta: Map<string, string> }> {
  const res = await fetch(`${BASE}/query`, {
    method: "POST",
    headers: { "content-type": "application/json", authorization: `Bearer ${KEY}` },
    body: JSON.stringify(body),
  });
  if (!res.ok) {
    const err = await res.json().then((j) => j.error).catch(() => ({ code: "http_error", message: res.statusText }));
    const ra = res.headers.get("retry-after");
    throw new QueryError(res.status, err.code, err.message, ra ? Number(ra) : undefined);
  }
  const tables: Record<string, Table> = {};
  let meta = new Map<string, string>();
  try {
    // The body is several complete Arrow IPC streams back to back: one reader per stream.
    for await (const reader of RecordBatchReader.readAll(res.body!)) {
      const batches = [];
      for await (const batch of reader) batches.push(batch);
      meta = reader.schema.metadata;
      tables[meta.get("bl.row_type")!] = new Table(reader.schema, batches);
    }
  } catch (e) {
    throw new QueryError(0, "truncated", `response ended early: ${e}`);
  }
  // bl.objects (sent before any row) is how many rows the response carries.
  const received = Object.values(tables).reduce((n, t) => n + t.numRows, 0);
  if (received !== Number(meta.get("bl.objects"))) throw new QueryError(0, "truncated", `got ${received} of ${meta.get("bl.objects")} rows`);
  return { tables, meta };
}

import { query } from "./bloklake-arrow.ts";

const { tables, meta } = await query({
  fromBlock: 19426000, toBlock: 19426009,
  logs: [{ address: ["0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"] }],
  fields: { log: { blockNumber: true, data: true } },
});
const row = tables.logs!.get(0)!;
row.block_number;  // 19426000n: UInt64 columns are bigint in JavaScript
row.data;          // Uint8Array

Needs arrow (feature ipc), reqwest (feature json), serde_json and tokio.

Cargo.toml

[dependencies]
arrow = { version = "59", default-features = false, features = ["ipc"] }
reqwest = { version = "0.13", features = ["json"] }
serde_json = "1"
tokio = { version = "1", features = ["full"] }

src/bloklake_arrow.rs

use arrow::ipc::reader::StreamReader;
use arrow::record_batch::RecordBatch;
use std::collections::HashMap;
use std::io::Cursor;

type Error = Box<dyn std::error::Error + Send + Sync>;

/// The decoded response: record batches per row type, and the `bl.*` metadata.
pub struct Response {
    pub tables: HashMap<String, Vec<RecordBatch>>,
    pub meta: HashMap<String, String>,
}

/// POST /query and decode every Arrow stream of the response.
pub async fn query(client: &reqwest::Client, base: &str, key: &str, body: &serde_json::Value) -> Result<Response, Error> {
    let resp = client.post(format!("{base}/query")).bearer_auth(key).json(body).send().await?;
    let status = resp.status();
    if !status.is_success() {
        return Err(format!("{status}: {}", resp.text().await?).into());
    }
    // A response carries about 8 MB of row data at most, so buffering it is fine.
    let mut cursor = Cursor::new(resp.bytes().await?);
    let mut tables = HashMap::new();
    let meta = loop {
        // One complete IPC stream; the reader stops at its end-of-stream marker.
        let reader = StreamReader::try_new(&mut cursor, None)?;
        let meta = reader.schema().metadata().clone();
        let batches = reader.collect::<Result<Vec<_>, _>>()?;
        tables.insert(meta["bl.row_type"].clone(), batches);
        if tables.len() == meta["bl.streams"].split(',').count() {
            break meta;
        }
    };
    // bl.objects (sent before any row) is how many rows the response carries.
    let received: usize = tables.values().flatten().map(|b| b.num_rows()).sum();
    if received.to_string() != meta["bl.objects"] {
        return Err(format!("truncated response: got {received} of {} rows", meta["bl.objects"]).into());
    }
    Ok(Response { tables, meta })
}

In any other language, the recipe is the same: open an Arrow IPC stream reader on the response body, read it to the end, read the next one from the same position, repeat bl.streams times, then compare the row count with bl.objects. Go’s ipc.NewReader, Java’s ArrowStreamReader and C++‘s RecordBatchStreamReader all work this way.

Into a dataframe

Arrow tables convert to dataframes without copying the data.

import polars as pl

logs = pl.from_arrow(tables["logs"])
# topics are 32 bytes; an address in a topic is the last 20
logs = logs.with_columns(
    ("0x" + pl.col("topic1").bin.slice(12).bin.encode("hex")).alias("from"),
    ("0x" + pl.col("topic2").bin.slice(12).bin.encode("hex")).alias("to"),
)

df = tables["logs"].to_pandas()
df["from"] = df["topic1"].map(lambda b: "0x" + b[12:].hex())

import duckdb

txs = tables["transactions"]           # DuckDB reads Arrow tables in place
duckdb.sql("select count(*), sum(gas_used) from txs").fetchall()

See Data types for how hashes, addresses and 128/256-bit amounts are encoded, with helpers to convert them.

Reading incrementally

The readers above hold one response in memory at a time. A response carries about 8 MB of row data at most, so that’s fine for almost everyone, and the pagination loop keeps memory flat across a whole backfill. If you want to process rows while a response is still arriving (lower latency to the first row), iterate the record batches instead of collecting them (for batch in reader in pyarrow, for await (const batch of reader) in arrow-js). Keep reading promptly: a response that isn’t read for 60 seconds, or is read slower than 50 KB/s, is cut off.

Get started

Chains

Querying

Responses

Guides

Account

Billing

Production

The response layout

Complete readers

Into a dataframe

Reading incrementally

​The response layout

​Complete readers

​Into a dataframe

​Reading incrementally

The response layout

Complete readers

Into a dataframe

Reading incrementally