blockdb API

blockdb 负责存取。日常写索引管道大多数时候只需要 Pipeline，这一页给需要直接操作数据层、或者要绕开 Pipeline 做精细控制的场景。

Table —— 普通表

不带区块语义的读写接口，高吞吐。建表本身走控制台页面，SDK 不提供建表 API，只能引用已经存在的表：

from blockdb import Table

t = Table("myspace.rules")

# 写：upsert_rows，sync=True 等服务端确认写入完成
t.upsert_rows([
    {"id": "rule_1", "threshold": 100},
    {"id": "rule_2", "threshold": 999},
], sync=True)

# 读
t.get_row("rule_1")                       # 按主键单行
t.batch_get_rows(["rule_1", "rule_2"])    # 按主键批量
t.scan("SELECT * FROM myspace.rules")     # 整表拉回 list[dict]，小表用；大表会全量进内存
t.filter_rows(filters="threshold > 100", order_by="threshold desc", limit=10)   # 按非主键条件查少量行

条件写入（幂等）

upsert_rows 支持 condition，只在满足条件时才覆盖已有行——常用于”后到的旧数据不应该覆盖已有的新值”这类幂等写入：

t.upsert_rows(
    [{"id": row_id, "first_seen_at": ts}],
    condition=("first_seen_at", "if_smaller"),   # 只有新值更小才覆盖
)

BlockTable —— Block Event / Block State 表

按区块高度读写，读取自动限制在 height <= 目标高度，不会读到还没确认的数据。

from blockdb import BlockTable, Block

t = BlockTable("token.token_transfer.eth")

# Block = 区块的对外标识
block = Block(id="0x...", height=25_015_000, timestamp=1_717_000_000)

t.get_block_rows(block)                # 取某个高度整块的所有行
t.upsert_block(block, rows)            # 按区块原子写入
t.upsert_block_bundle(bundle_number, rows)   # 按 bundle（1000 块一组）原子写入一批行

# Block State 表专属：按锚点查状态 / 事件快照
t.get_state(id, current_block=block)   # 取 <= 该高度的最新快照
t.get_event(id, current_block=block)   # 事件的 height 必须 <= current_block.height 才可见

平台会为每张 Block 表自动维护几张辅助表：

辅助表	内容
`<table>._height`	处理进度——每行是一段已完成的连续高度区间
`<table>._archive`（仅 Block State）	历史高度的快照，主键 `(id, height)`
`<table>._write`	批量写入（回填）任务的执行状态
`<table>._bundle`	每 1000 块一组的数据哈希，内部同步用

排查数据完整性、判断有没有写完，直接看 _height 就够了——不需要在代码里操作它。

TimeTable —— 按时间分桶

按固定时间桶（分钟级）组织的数据，写入时间自动向后取整到整分钟：

from blockdb import TimeTable

t = TimeTable("myspace.prices")
t.upsert_rows(rows)                        # 写：按时间桶 upsert
t.get_row(id)                               # 读最新值
t.get_row(id, time_at="2026-08-19T09:00:00Z")   # 读某个时刻的切面值（按桶对齐）

时间统一按 RFC3339 字符串处理：写入时可以传比较随意的格式（如 "2026-05-18 17:04:35"，不带时区按 UTC 理解），读回来的字段固定是 ...Z 形态。

如果你用的是 Pipeline 的 update()，订阅已经被包好了，通常不需要直接用 Subscribe。

每次一张表成功写入一个新块，平台会产出一个写入事件。Subscribe.listen() 是无限生成器，逐个 yield (表名, Block)，断线自动重连：

from blockdb import BlockTable, Subscribe

trace = BlockTable("chain.trace.eth")

for table_id, block in Subscribe(tables=[trace]).listen():
    rows = trace.get_block_rows(block)
    ...

# 从过去某个时间点接着收：start_at 收 RFC3339 字符串或 epoch 秒，不填 = 从最新开始
sub = Subscribe(tables=[trace], start_at="2026-08-19T09:00:00Z")
sub.close()   # 可以从别的线程调用，叫停卡在 listen() 上的循环

只有逐块写入（upsert_block）才会发订阅事件，批量写入（bundle / 回填）按设计不发事件。 上游用回填补的数据，下游订阅是收不到通知的，需要下游也主动发起一次回填。

Subscribe 只能订阅 Block Event / Block State 表，Normal 表没有订阅能力。

Aligned —— 多表高度对齐

转换逻辑经常需要点查另一张表的当前状态。如果那张表还没处理到当前高度，查到的就是过期数据——Aligned 是用来解决这个问题的状态机。

from blockdb import BlockTable, Subscribe, Aligned

event = BlockTable("chain.event.eth")   # trigger：要消费的源
token = BlockTable("token.token.eth")   # depends：必须已追平到该高度

sub = Subscribe(tables=[event, token])
aligned = Aligned(sub, loose_align=[event], strict_align=[token])

for block in aligned.listen():
    # 此刻 token 的共识高度 >= block.height，函数内查 token.token.eth 一定是追平的状态
    ...

两种角色：

角色	规则
trigger 表（`loose_align`）	该高度有数据即可放行
depends 表（`strict_align`）	该高度及之前所有高度都必须已处理完

只有全部 trigger 表在该高度 ready、且全部 depends 表的共识高度都 ≥ 该高度，这个高度才会被放行，且每个高度只放行一次。一张表可以同时是 trigger 又是 depends——两个位置都要写，写一边不会顶替另一边（需要 blockdb-py >= v0.1.23）。

共识高度 = 表的 _height 子表里第一个连续区间的右边界（不要求从 0 开始）；一行都还没有时取 start_height - 1。这也是为什么驱动逻辑不能跳过高度——跳过的高度会在 _height 里留下一个洞，后面所有更高的高度都会被卡住。

如果用的是 Pipeline，把依赖表传给 depends 参数即可，对齐由它自动处理，不需要手写 Aligned。

scan vs filter_rows

方法	行为	适用
`scan(sql)`	整表一次拉回 `list[dict]`，自己写 `FROM ...`	小表、需要在本地跑复杂计算
`filter_rows(filters, order_by, limit)`	按非主键条件查少量行，`filters` 是 SQL-like 字符串	大表按条件取少量行

Python SDK

参考

Table —— 普通表

条件写入（幂等）

BlockTable —— Block Event / Block State 表

TimeTable —— 按时间分桶

Aligned —— 多表高度对齐

scan vs filter_rows

​Table —— 普通表

​条件写入（幂等）

​BlockTable —— Block Event / Block State 表

​TimeTable —— 按时间分桶

​Subscribe —— 订阅新块

​Aligned —— 多表高度对齐

​scan vs filter_rows

Table —— 普通表

条件写入（幂等）

BlockTable —— Block Event / Block State 表

TimeTable —— 按时间分桶

Subscribe —— 订阅新块

Aligned —— 多表高度对齐

scan vs filter_rows