blockx API

blockx 负责算——把「源表 + 函数 + 目标表」打包成任务提交给远端 worker。日常写索引管道用 Pipeline 就够了；这一页是 Pipeline 内部实际在用的底层 API，给需要绕开 Pipeline、做精细控制的场景。

心智模型

一个任务 = 一个 CallConfig（算什么）+ 一个 ResultHandler（结果怎么处理）。Pipeline 就是把这一整套包起来的高层封装。

Trigger 与 func 的三种给法

from blockx import Trigger, SOURCE_ROW

Trigger(
    table="chain.trace.eth",
    func=...,                        # 见下方三种给法
    params=[SOURCE_ROW, "eth"],      # SOURCE_ROW 占位符：这个位置传源表的那一行
    operator=...,                    # 可选：执行前的过滤 / 去重管道
)

写法	行为
`func=callable`（一个 Python 函数）	SDK 用 `inspect.getsource` 抓源码，内联传给 worker——日常首选，直接写 `def`
`func="<space>.<函数名>"`（字符串）	引用一个已经在「Function 详情页」保存的 Function 资源，按 ID 在后端执行
`func=Function(source_code="...")`	显式持有一段源码字符串，想把代码当数据管理时用

params 里除了 SOURCE_ROW，其余原样作为函数的位置参数——只有位置参数，没有命名参数，这是跨语言设计决定的。

CallConfig（现有三种）

CallConfig	喂数据方式	单 task 粒度	场景
`BlockTableCallConfig`	订阅区块流，按 block 喂	一个 block	实时
`CallListCallConfig`	自己准备一批 callList（每行一组位置参数）	一段 callList	批处理 / dry-run
`BlockBundleCallConfig`	按 bundle（=1000 block），数据源来自 S3	一个 bundle	历史回填

BlockTableCallConfig(
    block=Block(...),
    triggerSources=[dict(table=源表, func=业务函数, params=[SOURCE_ROW, "eth"])],
)
CallListCallConfig(
    func=业务函数,
    callList=[[record, "eth"], ...],   # 每行 = 一组位置参数
)

ResultHandler（现有四种）

ResultHandler	用途
`BlockWriteHandler(targetTable, block)`	落库到 Block 表
`ReturnValueResultHandler()`	只回传函数返回值，不落库（dry-run）
`BlockBundleWriteResultHandler(targetTable, number)`	bundle 回填落库
`TableUpsertsResultHandler(targetTable)`	落 Normal 表

组装 + 提交

from blockx import TaskBuilder

task = TaskBuilder.build(call_config=call_config, handler=handler)
result = task.submit(timeout=60.0)

if result.task_result.success:
    n = (result.handler_result or {}).get("written_rows", 0)
else:
    print(result.task_result.failure_code, result.error)

success 只代表 worker 受理了这个任务，写库是后台异步做的——这一刻数据不一定已经进表。真实进度看目标表的 _write 子表。

task.submit() 常用参数：

参数	默认	说明
`timeout`	`60.0`	整体等待上限（秒），超时抛 `TimeoutError`
`poll_interval`	`0.2`	轮询结果的间隔（秒），仅在推送不可用时降级生效

Function 互调用

一个 Function 可以在执行过程中调用另一个已保存的 Function：

import blockx.function as function

result = function.call(function_id, arg1, arg2, timeout=5)

在 worker 里跑走同一个 task 内的子调用，在本地 / Notebook 里跑走 sync-invoker。

Operator：执行前预过滤 / 去重

有些过滤、去重逻辑不必塞进业务函数——可以在 Trigger 上挂一条 operator 管道，引擎会在把行喂给函数之前先做掉：

from blockdb import Column
from blockdb.operator import filter as op_filter

op = op_filter((Column("value") > 0) & (Column("name") == "Transfer")).deduplicate(Column("tx_id"))

Trigger(table="chain.trace.eth", func=to_transfer, params=[SOURCE_ROW, "eth"], operator=op)

operator 接受 Operator 对象、dict，或者它们的 JSON 字符串，三种写法效果一样。

调试：不落库预演

联调新转换逻辑时，用环境变量开启 debug 模式——worker 端不会真正写目标表，读 / 算逻辑照常跑，只有最后一步落库被替换成打日志：

export BLOCKDB_DEBUG=1

这个开关和 blockdb-py 共用，设一次就把整条 SDK 链路一起切到 debug，代码不用改。上线前记得关掉，否则生产任务的写入会全部进日志而不落库。

Python SDK

参考

心智模型

Trigger 与 func 的三种给法

CallConfig（现有三种）

ResultHandler（现有四种）

组装 + 提交

Function 互调用

Operator：执行前预过滤 / 去重

调试：不落库预演

​心智模型

​Trigger 与 func 的三种给法

​CallConfig（现有三种）

​ResultHandler（现有四种）

​组装 + 提交

​Function 互调用

​Operator：执行前预过滤 / 去重

​调试：不落库预演

心智模型

Trigger 与 func 的三种给法

CallConfig（现有三种）

ResultHandler（现有四种）

组装 + 提交

Function 互调用

Operator：执行前预过滤 / 去重

调试：不落库预演