> ## Documentation Index
> Fetch the complete documentation index at: https://docs.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.

# blockdb API

> 数据存取层的完整参考：Table、BlockTable、TimeTable、Subscribe、Aligned。

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

## Table —— 普通表

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

```python theme={null}
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`，只在满足条件时才覆盖已有行——常用于"后到的旧数据不应该覆盖已有的新值"这类幂等写入：

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

## BlockTable —— Block Event / Block State 表

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

```python theme={null}
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 —— 按时间分桶

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

```python theme={null}
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")   # 读某个时刻的切面值（按桶对齐）
```

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

## Subscribe —— 订阅新块

<Tip>
  如果你用的是 [Pipeline](/indexing/run/pipeline) 的 `update()`，订阅已经被包好了，通常不需要直接用 `Subscribe`。
</Tip>

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

```python theme={null}
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() 上的循环
```

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

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

## Aligned —— 多表高度对齐

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

```python theme={null}
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）。

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

如果用的是 [Pipeline](/indexing/run/pipeline)，把依赖表传给 `depends` 参数即可，对齐由它自动处理，不需要手写 `Aligned`。

## scan vs filter\_rows

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