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

# 读写数据

> 使用 Table、BlockTable、TimeTable 三类表模型封装进行数据读写。

# 读写数据

BlockDB 的 Python SDK 把三类逻辑表封装成对应的类：普通表 `Table`、Block 表 `BlockTable`、Time 表 `TimeTable`。每个类都是薄封装，方法最终委托到底层 gRPC，查询方法返回 Python `dict` 或 `list[dict]`，写入方法返回 `job_id`。

## 三类表模型与选型

| 类 | 对应逻辑表 | 数据维度 | 读锚点 |
| - | - | - | - |
| `Table(name)` | 普通表 | 行级 | 主键 `row_id` |
| `BlockTable(name)` | Block State / Event 表 | 区块高度 | `current_block` / `block` |
| `TimeTable(name)` | Time 表 | 物理时间桶 | `time_at` |

选择哪种表模型，取决于你的数据形态：普通表处理行级数据，Block 表处理随区块高度的状态与事件，Time 表处理随物理时间变化的指标。详见[数据模型](/zh/concepts/data-model)。

## 普通表（Table）

针对行级数据，按主键读写与全表扫描：

```python theme={null}
from blockdb import Table

t = Table("demo.event_log")

# 写：实时写入
job_id = t.upsert_rows([{"id": "evt_001", "name": "login", "value": 1}])

# 读：单行 / 批量 / 条件筛选
row = t.get_row("evt_001")
rows = t.batch_get_rows(["evt_001", "evt_002"])
hits = t.filter_rows(filters="value > 0", order_by="value desc", limit=100)

# 扫描：全表 / 流式，返回 list[dict]
all_rows = t.scan("SELECT * FROM demo.event_log")
for row in t.scan_iter("SELECT * FROM demo.event_log WHERE name = 'login'"):
    ...  # 大表逐行 yield，内存恒定

# 删除
t.delete_rows(["evt_001"])
```

<Info>
  大表检索请用 `scan_iter`（流式逐行），避免 `list(scan_iter(...))` 一次性吃光内存。传入的 SQL 以流式方式执行，可处理百万级结果集。
</Info>

## Block 表（BlockTable）

针对区块维度的状态与事件，按区块锚点读写：

```python theme={null}
from blockdb import BlockTable, Block

bt = BlockTable("token.token.eth")
block = Block(id="0xabc...", height=18000000, timestamp=1712345678)

# 读：按区块时刻读状态 / 事件 / 某一高度整块数据
state = bt.get_state("acct_1", block)
event = bt.get_event("evt_1", block)
rows = bt.get_block_rows(block)

# 写：按区块原子写入
bt.upsert_block(block, [{"id": "acct_1", "balance": 1000}])
```

`get_state` / `get_event` 会以 `current_block` 为锚点，自动注入 `height ≤ 目标高度`，返回最近一次不超过该高度的快照。当前区块高度需不超过表的共识高度才能读到正确结果，详见[Block State 与 Block Event 表](/zh/concepts/block-tables)。

## Time 表（TimeTable）

针对随物理时间变化的指标，按时间桶读写：

```python theme={null}
from blockdb import TimeTable

tt = TimeTable("token_time_demo")

# 写：往时间桶批量写入（time_at 自动向后取整到整分钟）
tt.upsert_time_rows("2026-05-18 17:04:35", [{"id": "0xabc...", "value": 2543.12}])

# 读：某个对象在某个时间桶的状态
row = tt.get_value("0xabc...", "2026-05-18 17:04:40")
```

`TimeTable` 在 SDK 层自动完成 60 秒时间桶对齐：`17:04:35` 对齐到 `17:05:00`，读写口径一致。详见[Time 表](/zh/concepts/time-table)。

## 进阶写入

* **条件更新**：普通表 `upsert_rows` 支持 `condition=(列名, "if_smaller"/"if_larger")`，仅当新值更小/更大时才更新，适合余额、高度等单调值：

  ```python theme={null}
  t.upsert_rows(rows, condition=("value", "if_larger"))
  ```

* **批量写**：百万行级别的回填走 S3 中转的批量写接口，返回 `job_id`，可查询状态或取消。详见[写入架构](/zh/architecture/batch-write)。

## 相关阅读

* [API 总览](/zh/reference/api)
* [实时订阅与多表对齐](/zh/guides/subscribe)
* [条件筛选与算子管道](/zh/guides/filter-operator)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.