blockdb-py

从 blockdb 包导入公开接口。在 Chaintable Notebook 中使用写入和订阅；Function 中可使用允许的读取接口。源码提供完整实现。

普通表

`NormalTable(table_id)`

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

方法	参数	返回值
`get(row_id)`	主键	记录字典，缺失时为 `None`
`get_many(row_ids)`	主键列表	按输入顺序返回记录列表，缺失位置为 `None`
`filter(expr)`	列表达式	待执行的查询对象
`select(*columns)`	要读取的列	待执行的查询对象
`query(offset=0, limit=0, order_by='')`	偏移量、条数和 SQL 排序	记录列表
`find(filters='', order_by='', limit=0, offset=0)`	SQL 条件、排序、条数和偏移量	记录列表
`scan(sql=None)`	可选的完整 SQL 语句，省略时扫描整张表	全部结果的列表
`scan_iter(sql=None)`	与 `scan()` 相同	逐行返回记录的迭代器
`write(rows, condition=None)`	记录列表、可选更新条件	请求被接受后返回 `True`
`batch_write(rows, condition=None)`	批量记录、可选更新条件	异步任务提交后返回 `True`
`delete(row_ids)`	要删除的主键列表	请求被接受后返回 `True`

from blockdb import C, NormalTable

notes = NormalTable('demo.asset_notes')
print(notes.get('usdc'))
print(notes.filter(C.symbol == 'USDC').query(limit=5))
print(notes.select(C.id, C.symbol).scan())

先按创建和管理表建立示例表，再按读写数据写入记录。 filter() 和 select() 只收集条件和列，query()、scan() 或 scan_iter() 才真正读取，同一个查询对象可以多次执行。query() 是在线读取，支持排序和分页，始终返回全部列；scan() 用于读取大量记录，不支持排序和分页，但 select() 对它生效。 find() 接受手写的 SQL 条件，不包含 WHERE 关键字，用于列表达式还无法表达的条件。scan() 传入完整语句时按原样执行，FROM 由调用方书写，并用反引号引用完整表 ID；SDK 不会根据对象自动补全。它是 BlockDB 扫描接口，支持的 SQL 范围与分析引擎不同。

条件写入

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() 每行需要相同的字段集合。返回成功不保证随后立即读到新数据；批量写入的状态检查见查看异步写任务。

区块事件表与状态表

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

table_id 为表 ID，block_id 指定读取位置。事件表按事件 ID 读取记录；状态表按业务 ID 读取该区块时的状态。省略 block_id 时读取每个 ID 的最新记录，不会退回共识高度；显式指定区块时，服务端检查表的共识高度是否已覆盖该区块。

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

from blockdb import StateTable

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

本例读取最新状态，示例表来自追踪流动性池储备变化。按指定区块读取状态时，表的共识高度须覆盖该区块，详见工作原理。 get_block_rows() 的 None 和 [] 含义不同：[] 表示该区块已处理且没有记录，None 表示链上没有该区块，或该表还没处理到这个高度。逐块 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`

from blockdb import TimeTable

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

write() 使用方法参数的时间，不使用行内 time_at；多时刻回填使用 batch_write()，每行只包含固定的三个字段。时间分桶由服务端处理，同一桶保留最新时间点，分辨率随数据年龄变化，详见工作原理。

区块订阅

接口	参数与返回值
`Subscribe(tables=None, requires=None, lookback=100, readiness=RowsReady)`	触发表、依赖表、回放区块数和就绪判定
`sub.listen()`	持续产生 `WorldStateReadyEvent`，同一高度只产生一次
`sub.listen_events()`	未对齐的原始流，产生 `(表 ID, 事件, 类型)`
`sub.close()`	停止订阅，可从其他线程调用
`WorldStateReadyEvent`	字段为 `block_id`、`block_height` 和 `block_timestamp`
`Committed` / `RowsReady`	`readiness` 的两个取值

对齐内置在订阅中：tables 中每张表都写入该高度、且 requires 中每张表的共识高度到达该高度后，listen() 才产生事件。要求互不对齐的表应分别订阅。tables 和 requires 都接受单个表对象、表 ID 或它们的列表。 lookback 是回放的区块数，含链头本身：100 从链头往前第 100 个区块开始，0 只接收新事件。服务端按每张表所在链解析该窗口，超出回放窗口的请求会被拒绝，历史数据仍需回填。 readiness 默认 RowsReady，区块的记录可读时就计入对齐；Committed 只接受已提交的通知。依赖表始终按已提交判定。完整运行示例见订阅区块事件。

筛选表达式和类型

from blockdb import Column
from blockdb.operator import filter

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

Column 和 filter() 构造 Pipeline 触发源的筛选表达式；逻辑组合使用带括号的 &、|，集合条件使用 in_() 和 not_in()，接在 filter() 后的 dedup(*cols) 按列去重。C 是列的简写，C.name 等价于 Column('name')。筛选作用于源记录的原始值：UINT256 列按十进制字符串比较，常量需写成字符串，例如 Column('amount') == '1500000'；大小比较按字符串顺序而非数值大小。 filter() 会遮蔽内置的 filter，因此从 blockdb.operator 导入；表上的 filter() 是查询方法，不需要导入。不要把筛选表达式传给要求 SQL 条件字符串的 find()。 LogicalType 提供类型名称，restore_value(logical_type, raw) 按元数据把原始值恢复为 Python 值。例如 restore_value('UINT256', '1500000') 保留十进制字符串，计算时再使用 int()。类型说明见数据类型。

概览

快速开始

使用指南

参考

普通表

`NormalTable(table_id)`

条件写入

区块事件表与状态表

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

时间表

`TimeTable(table_id, time_at=None)`

区块订阅

筛选表达式和类型

普通表

NormalTable(table_id)

条件写入

区块事件表与状态表

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

时间表

TimeTable(table_id, time_at=None)

区块订阅

筛选表达式和类型

`NormalTable(table_id)`

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

`TimeTable(table_id, time_at=None)`