blockx-py

BlockX 将计算函数与触发、输入和结果写入组合起来。以下驱动接口默认在 Notebook 中使用；函数内复用逻辑可使用 function.call()。源码提供完整实现和低层接口。

同步调用 Function

`function.call(function_id, *args, timeout=5)`

from blockx import function

amount = function.call('demo.usdc_amount', 1500000)
print(amount)

function_id 为已保存的 Function ID，*args 按参数顺序原样传入，不按 Function 声明的参数类型转换；示例返回 1.5，前提是已按创建 Function创建示例函数。 timeout 单位为秒，默认 5 秒，也是允许的最大值，超过会被拒绝。在 Notebook 中可以设得更短；在 Function 内部调用其他函数时，受父调用的剩余期限限制，不能通过增加子调用超时来延长整次执行。失败时抛出 FunctionInvokeError，常用属性为 code、message、retryable、call_id 和 origin。origin 说明失败归属：FAILURE_ORIGIN_USER 表示函数自身的代码出错，FAILURE_ORIGIN_CAPACITY 表示当前没有可用算力，FAILURE_ORIGIN_SYSTEM 表示平台侧失败。

Function 描述对象

`Function(source_code=None, function_id='', name='', mode='inline')`

参数	说明
`source_code`	内联 Python 源码，入口为 `_`
`function_id`	已保存的 Function ID
`name`	可选名称
`mode`	当前支持 `inline`

至少提供源码或 Function ID。该对象用于任务配置，不会在构造时执行或保存函数。区块和区块包任务只接受已保存的 Function ID，内联源码仅 DirectCallConfig 接受。

Pipeline 与触发源

`Pipeline(sources, target, requires=None, condition=None, readiness=None)`

参数	说明
`sources`	触发源列表，至少一项；每项是一个字典，键见下表
`target`	目标表 ID 或表对象，其类型决定结果写入方式
`requires`	依赖表，不触发计算也不作为参数传入，但要求对应高度的数据先就绪
`condition`	写入普通表时的条件更新，其他目标表类型忽略
`readiness`	实时阶段的就绪判定，取值见 blockdb-py 的区块订阅

触发源字典的键：

键	说明
`table_id`	触发源表 ID 或表对象
`func_id`	已保存的 Function ID
`args_tmpl`	位置参数列表，使用 `ROW` 放置源记录；源记录中的 `UINT256` 字段以 Python `int` 传入，其他固定值按原样传入
`operator`	通过 `blockdb.operator` 的 `filter()` 构建的筛选表达式，可选

方法	参数	行为与返回值
`backfill(start, end=None, *, progress=None)`	包含两端的高度范围；省略结束高度时以可处理的上游进度为界	回填后返回 `BackfillResult`
`update(lookback=100)`	重新订阅时回放的区块数	从目标表的起始高度补齐缺口，然后持续处理新事件

新建的区块表未设置起始高度时，update() 会从高度 0 开始逐块补齐，耗时很长。只需要近期数据时先设置起始高度；需要历史数据时先用 backfill() 回填。同一张表可以同时出现在 sources 和 requires 中：作为触发源提供候选高度，作为依赖表要求该高度已处理完成。 BackfillResult 的常用属性：

属性	含义
`ok`	所有提交阶段是否成功
`done_blocks`	成功处理的区块数
`written_rows`	统计的写入行数
`failed_ranges`	失败区间列表，每项为 `(起始高度, 结束高度, 错误码, 信息)`

大范围回填的批量写入可能异步执行，应另外确认目标表的写任务和共识进度。可直接运行的完整配置见构建 Pipeline。

自定义任务

`DirectCallConfig(func_id=None, args_list=None)`

func_id 接受 Function ID、callable 或 Function 对象；args_list 是位置参数列表的列表，每个子列表触发一次调用。

from blockx import DirectCallConfig, ReturnValueHandler, TaskBuilder

config = DirectCallConfig(
    func_id='demo.usdc_amount',
    args_list=[[1500000], [2000000]],
)
task = TaskBuilder.build(config, ReturnValueHandler())
result = task.submit(timeout=30)
if not result.task_result.success:
    raise RuntimeError(result.error)
print(sorted(result.handler_result))

成功时输出 [1.5, 2.0]。原始结果顺序不保证对应输入顺序，需要关联时在返回值中携带业务 ID。

`TaskBuilder.build(call_config, handler)`

返回尚未提交的 Task。task.submit(timeout=None) 提交并等待，返回 TaskResult；timeout 单位为秒，省略时使用服务端默认值。异步 Python 代码可使用 await task.submit_async(timeout=...)。

`TaskResult` 属性	含义
`task_result.success`	任务是否成功
`task_result.failure_code`	失败代码
`task_result.retryable`	是否标记为可重试
`handler_result`	结果处理器返回的内容
`error`	错误标识，成功时通常为 `None`

区块任务输入

配置对象	参数与用途
`BlockCallConfig(block_id=None, sources=None)`	按一个区块读取源表，`block_id` 为区块 ID 字符串
`BundleCallConfig(bundle_id=None, sources=None)`	按区块包读取，`bundle_id` 为区块包编号
`BundleCallConfigTemplate(sources=None)`	只填好触发源；`for_bundle(n)` 生成对应区块包的配置

sources 与 Pipeline 的触发源使用同一组键，ROW 表示源记录所在的参数位置。

结果处理器

处理器	参数	写入约定
`ReturnValueHandler()`	无	收集函数返回值并交回 Notebook
`NormalTableWriteHandler(target, condition=None)`	目标普通表、可选条件	返回记录应包含主键及业务字段
`TimeTableWriteHandler(target)`	目标时间表	返回记录包含 `id`、`time_at`、`value`，同一任务共用一个 `time_at`
`BlockTableWriteHandler(target)`	目标区块表	写入结果及对应区块进度；返回记录需包含 `block_id`、`block_height`、`block_timestamp`

区块处理器不自带区块或区块包，写入位置取自同一个任务的输入配置，因此两者不会不一致；它需要配合 BlockCallConfig 或 BundleCallConfig 使用。标准 Pipeline 根据目标表类型选择处理器。逐块写入时，服务端按区块补全三个区块字段；按区块包写入时不会补全，缺少这些字段的区块包会写入失败，而回填结果仍显示成功。backfill() 中覆盖完整区块包的部分和 backfill_block_bundles() 都按区块包写入，因此函数应从源记录复制这三个字段一并返回。

回填辅助接口

backfill_block_bundles(start_bundle_id, end_bundle_id, config, handler, *, retry=2, timeout=None, progress=None, abort_if_first_fails=True)

用于自定义区块包回填：config 传入 BundleCallConfigTemplate，handler 传入写入处理器，范围包含两端，返回 BackfillSummary。

`BackfillSummary` 属性	含义
`ok`	是否没有失败的区块包
`total` / `succeeded` / `failed`	本次区块包总数、成功数和失败数
`failed_bundles`	失败的区块包编号列表
`written_rows`	统计的写入行数
`wall_seconds`	本次运行耗时

retry 是每个区块包就地重试的次数；首个区块包重试后仍失败时默认中止整次运行，多数情况下这是配置问题。省略 progress 时按完成比例打印进度，传入 callable 可自行处理。 bundle_height_range(bundle_id) 返回区块包包含的高度范围。低层配置和测试能力 LocalTestService 见 GitHub。

概览

快速开始

使用指南

参考