> ## 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-py

> 普通表、区块表、时间表、区块订阅和筛选条件的 Python 接口参考。

从 `blockdb` 包导入公开接口。在 Chaintable Notebook 中使用写入和订阅；Function 中可使用允许的读取接口。[源码](https://github.com/Chaintable/blockdb-py)提供完整实现。

## 普通表

### `NormalTable(table_id)`

`table_id` 是完整表 ID，返回普通表对象。

| 方法                                                        | 参数               | 返回值                      |
| --------------------------------------------------------- | ---------------- | ------------------------ |
| `get(row_id)`                                             | 主键               | 记录字典，缺失时为 `None`         |
| `get_many(row_ids)`                                       | 主键列表             | 按输入顺序返回记录列表，缺失位置为 `None` |
| `filter_rows(filters='', order_by='', limit=0, offset=0)` | SQL 条件、排序、条数和偏移量 | 记录列表                     |
| `scan(sql)`                                               | 完整的受支持 SQL 语句    | 全部结果的列表                  |
| `scan_iter(sql)`                                          | 与 `scan()` 相同    | 逐行返回记录的迭代器               |
| `write(rows, condition=None)`                             | 记录列表、可选更新条件      | 请求被接受后返回 `True`          |
| `batch_write(rows, condition=None)`                       | 批量记录、可选更新条件      | 异步任务提交后返回 `True`         |
| `delete(row_ids)`                                         | 要删除的主键列表         | 请求被接受后返回 `True`          |

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

notes = NormalTable('demo.asset_notes')
print(notes.get('usdc'))
print(notes.filter_rows("symbol = 'USDC'", limit=5))
print(notes.scan("SELECT id, symbol FROM `demo.asset_notes` WHERE id = 'usdc'"))
```

先按[创建和管理表](/zh/guides/tables/create-and-manage)建立示例表，再按[读写数据](/zh/guides/tables/read-and-write)写入记录。

`filter_rows()` 的条件不包含 `WHERE` 关键字，可通过 `limit` 限制条数。`scan()` 中使用反引号引用完整表 ID；SDK 不会根据对象自动补全 `FROM`。它是 BlockDB 扫描接口，支持的 SQL 范围与分析引擎不同。

## 条件写入

```python theme={null}
from blockdb import NormalTable, IF_LARGER

NormalTable('demo.asset_notes').write(
    [{'id': 'usdc', 'symbol': 'USDC', 'amount': 2000000}],
    condition=('amount', IF_LARGER),
)
```

`condition` 为 `(列名, 模式)`：`IF_LARGER` 仅在新值较大时更新已有记录，`IF_SMALLER` 仅在新值较小时更新；不存在的记录仍会插入。条件列不能是主键，同一批次共用一个条件。

`batch_write()` 每行需要相同的字段集合。返回成功不保证随后立即读到新数据；批量写入的状态检查见[查看异步写任务](/zh/guides/tables/advanced#查看异步写任务)。

## 区块事件表与状态表

### `EventTable(table_id, block_id=None)` / `StateTable(table_id, block_id=None)`

`table_id` 为表 ID，`block_id` 指定读取位置。事件表按事件 ID 读取记录；状态表按业务 ID 读取该区块时的状态。省略 `block_id` 时，状态表直接读取该对象的 latest，事件表读取该事件 ID 在最高区块的记录，不受共识高度限制。显式指定区块时，服务端检查表的共识高度是否已覆盖该区块。

| 方法                             | 参数                  | 返回值             |
| ------------------------------ | ------------------- | --------------- |
| `get(row_id)`                  | 事件 ID 或状态 ID        | 记录字典或 `None`    |
| `get_many(row_ids)`            | ID 列表               | 与输入对应的记录列表      |
| `get_block_rows(block)`        | `Block` 对象或含区块信息的对象 | 该区块写入的记录列表      |
| `get_consensus_height()`       | 无                   | 表共识高度，整数        |
| `write(rows, block_id)`        | 结果行和区块 ID           | 请求被接受后为 `True`  |
| `batch_write(rows, bundle_id)` | 含区块字段的结果行和区块批次编号    | 异步任务提交后为 `True` |

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

reserves = StateTable('demo.pool_reserves.eth')
print(reserves.get('0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc'))
```

本例读取最新状态，示例表来自[追踪流动性池储备变化](/zh/quickstart/build/track-pool-reserves)。按指定区块读取状态时，表的共识高度须覆盖该区块，详见[工作原理](/zh/guides/tables/model#区块数据与处理进度)。

逐块 `write()` 由服务端补齐区块字段；`batch_write()` 的每行必须携带 `block_id`、`block_height` 和 `block_timestamp`。区块批次编号不是区块高度。

## 时间表

### `TimeTable(table_id, time_at=None)`

`time_at` 指定读取时刻；省略时读取最新值。可使用 RFC3339 时间字符串或 Python `datetime`，无时区时间按 UTC 解释。

| 方法                     | 参数                          | 返回值                                  |
| ---------------------- | --------------------------- | ------------------------------------ |
| `get(row_id)`          | 业务 ID                       | 含 `id`、`time_at`、`value` 的记录或 `None` |
| `get_many(row_ids)`    | ID 列表                       | 按输入顺序返回记录，缺失为 `None`                 |
| `write(rows, time_at)` | `{id, value}` 记录和所有行共用的时刻   | 请求被接受后为 `True`                       |
| `batch_write(rows)`    | 每行包含 `id`、`time_at`、`value` | 异步任务提交后为 `True`                      |

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

price = TimeTable('price.price.eth').get(
    '0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee'
)
print(price)
```

`write()` 使用方法参数的时间，不使用行内 `time_at`；多时刻回填使用 `batch_write()`，每行只包含固定的三个字段。

时间分桶由服务端处理，同一桶保留最新时间点，分辨率随数据年龄变化，详见[工作原理](/zh/guides/tables/model)。

## 区块与订阅

| 接口                                                  | 参数与返回值                   |
| --------------------------------------------------- | ------------------------ |
| `Block(id, height, timestamp)`                      | 区块 ID、整数高度、时间；返回区块对象     |
| `Subscribe(tables=None, start_at=None)`             | 表对象或表 ID 列表，可选事件时间游标     |
| `sub.listen()`                                      | 持续产生 `(table_id, Block)` |
| `sub.close()`                                       | 停止订阅，可从其他线程调用            |
| `Aligned(sub, loose_align=None, strict_align=None)` | 订阅对象、触发表列表、共识依赖表列表       |
| `aligned.listen()`                                  | 多表满足高度要求后产生 `Block`      |

`start_at` 是订阅事件的时间游标，接受 RFC3339 字符串或 Unix 秒数，不是历史区块高度。订阅重连不能保证补回任意久远的历史，应配合回填。

`loose_align` 至少有一张触发表；`strict_align` 要求依赖表的共识高度到达待处理区块。完整运行示例见[订阅区块事件](/zh/guides/notebooks/advanced#订阅区块事件)。

## 筛选表达式和类型

```python theme={null}
from blockdb import Column, filter

operator = filter(
    (Column('contract_id') == '0xb4e16d0168e52d35cacd2c6185b44281ec28c9dc')
    & (Column('name') == 'Sync')
)
```

`Column` 和 `filter()` 构造 Pipeline 触发源的筛选表达式；逻辑组合使用带括号的 `&`、`|`。不要把该表达式传给要求 SQL 条件字符串的 `filter_rows()`。

`LogicalType` 提供类型名称，`restore_value(logical_type, raw)` 按元数据把原始值恢复为 Python 值。例如 `restore_value('UINT256', '1500000')` 保留十进制字符串，计算时再使用 `int()`。类型说明见[数据类型](/zh/reference/data-types)。