BlockX-py 快速开始

from blockx import BlockTableCallConfig, BlockWriteHandler, TaskBuilder
from blockdb import Block

# 1) 写一个转换函数：入参是源表的一行 dict，返回 dict 就写入目标表，返回 None 就丢弃这行
#    worker 在隔离沙箱里执行它：常量 / 辅助逻辑都要写进函数体，不能引用函数外的全局名
def to_token_transfer(record, chain_id):
    value = float(record.get("value") or 0)        # FLOAT 列以字符串传入，先转一次再用
    if value <= 0 or not record.get("tx_id"):
        return None                                  # 不合格的行直接过滤掉
    return {
        "id":        record["id"],
        "from_addr": record["from_addr"],
        "to_addr":   record["to_addr"],
        "value":     value,
        "tx_id":     record["tx_id"],
    }

# 2) 指定要处理哪个区块
block = Block(id="0xblock...", height=25015000, timestamp=1717000000)

# 3) CallConfig：在这个 block 上、对源表的每一行跑 to_token_transfer
call_config = BlockTableCallConfig(
    block=block,
    triggerSources=[dict(
        table="chaintable.trace.eth",                 # 源表
        func=to_token_transfer,                        # 你的转换函数
        params=["${chaintable.trace.eth}", "eth"],     # ${表名} = 把该表本块的行逐行喂给函数
    )],
)

# 4) Handler：把函数返回的 dict 写到目标表
handler = BlockWriteHandler(targetTable="token.token_transfer.eth", block=block)

# 5) 组装 + 提交。task.submit() 自动跑完「派活 → 执行 → 等结果」整条链路
task = TaskBuilder.build(call_config=call_config, handler=handler)
result = task.submit()

print(result.task_result.success, result.handler_result)
# -> True {"written_rows": 42}

from blockx import BlockTableCallConfig, BlockWriteHandler, TaskBuilder
from blockdb import BlockTable
from blockdb.block_table import Subscribe        # L2 区块订阅（注意是 block_table 子包）

SOURCE = "chaintable.trace.eth"
TARGET = "token.token_transfer.eth"

def to_token_transfer(record, chain_id):
    value = float(record.get("value") or 0)
    if value <= 0 or not record.get("tx_id"):
        return None
    return {"id": record["id"], "from_addr": record["from_addr"],
            "to_addr": record["to_addr"], "value": value, "tx_id": record["tx_id"]}

src = BlockTable(SOURCE)

# listen() 是无限生成器，断连自动重连——直接 for 消费即可，不用自己写重试
for table_id, block in Subscribe(tables=[src]).listen():
    call_config = BlockTableCallConfig(
        block=block,
        triggerSources=[dict(
            table=src,
            func=to_token_transfer,
            params=[f"${{{SOURCE}}}", "eth"],
        )],
    )
    handler = BlockWriteHandler(targetTable=TARGET, block=block)
    task = TaskBuilder.build(call_config=call_config, handler=handler)

    res = task.submit(timeout=120.0)
    if res.task_result.success:
        n = (res.handler_result or {}).get("written_rows", 0)
        print(f"h={block.height} written={n}")
    else:
        print(f"h={block.height} error={res.error}")

from blockx import BlockTableCallConfig, BlockWriteHandler, TaskBuilder
from blockdb import BlockTable, Aligned
from blockdb.block_table import Subscribe

def event_to_token_transfer(record, chain_id):
    from blockdb import Table as _Table            # ★ worker 沙箱内重新 import

    if record.get("name") != "Transfer":
        return None
    contract_id = record.get("contract_id")

    # ★ worker 内 Online 表 lookup：查 token.token.<chain> 拿 decimals
    token = _Table(f"token.token.{chain_id}").get_row(contract_id)
    if not token:
        return None
    decimals = token.get("decimals")
    if decimals is None:
        return None

    import json
    params = json.loads(record["params"]) if isinstance(record.get("params"), str) else record.get("params")
    from_addr, to_addr, raw_amount = params
    return {
        "id": record["id"],
        "from_addr": from_addr.lower(),
        "to_addr": to_addr.lower(),
        "token_id": token["id"],
        "value": raw_amount / 10 ** decimals,
        "tx_id": record.get("tx_id") or "0x",
    }

event = BlockTable("chaintable.event.eth")        # trigger：我要消费的源
token = BlockTable("token.token.eth")             # depends：必须已追平到该高度

# 所有表（trigger + depends）都喂进同一个 Subscribe，再用 Aligned 包一层
sub = Subscribe(tables=[event, token])
aligned = Aligned(sub, loose_align=[event], strict_align=[token])

for block in aligned.listen():
    # 此刻 token 的共识高度 >= block.height，函数内查 token.token.eth 一定是追平的状态
    call_config = BlockTableCallConfig(
        block=block,
        triggerSources=[dict(
            table=event,
            func=event_to_token_transfer,
            params=["${chaintable.event.eth}", "eth"],
        )],
    )
    handler = BlockWriteHandler(targetTable="token.token_transfer.eth", block=block)
    TaskBuilder.build(call_config=call_config, handler=handler).submit()

from blockx import CallListCallConfig, ReturnValueResultHandler, BlockWriteHandler, TaskBuilder

# 业务函数与实时路径完全复用
def to_token_transfer(record, chain_id):
    value = float(record.get("value") or 0)
    if value <= 0 or not record.get("tx_id"):
        return None
    return {"id": record["id"], "value": value, "tx_id": record["tx_id"]}

# 假设 records 来自 blockdb 的历史扫描：Table("...").scan_iter("SELECT ...")
records = [...]

# callList：每行 = 一组位置参数 [record, chain_id]
calls = [[r, "eth"] for r in records]

# A) 只看结果、不落库（dry-run）：用 ReturnValueResultHandler
task = TaskBuilder.build(
    call_config=CallListCallConfig(func=to_token_transfer, callList=calls),
    handler=ReturnValueResultHandler(),
)
res = task.submit()
print(res.handler_result)        # 函数所有 return 值收集回来

# B) 直接落库：换成 BlockWriteHandler(targetTable=...) 即可
task = TaskBuilder.build(
    call_config=CallListCallConfig(func=to_token_transfer, callList=calls),
    handler=BlockWriteHandler(targetTable="token.token_transfer.eth"),
)
task.submit()

from blockdb import Column, BlockTable
from blockdb.operator import filter as op_filter
from blockx import BlockTableCallConfig, BlockWriteHandler, TaskBuilder

trace = BlockTable("chaintable.trace.eth")

# 管道：只保留 value > 0 的 Transfer 行，再按 tx_id 去重
op = (
    op_filter((Column("value") > 0) & (Column("name") == "Transfer"))
    .deduplicate(Column("tx_id"))
)

def to_token_transfer(record, chain_id):
    return {"id": record["id"], "value": float(record.get("value") or 0), "tx_id": record["tx_id"]}

call_config = BlockTableCallConfig(
    block=block,
    triggerSources=[dict(
        table=trace,
        func=to_token_transfer,
        params=[f"${{{trace.name}}}", "eth"],
        operator=op.to_json(),          # ★ 挂上预处理管道（str 或 dict 均可）
    )],
)
handler = BlockWriteHandler(targetTable="token.token_transfer.eth", block=block)
TaskBuilder.build(call_config=call_config, handler=handler).submit()

export BLOCKDB_DEBUG=1     # 1 / true / yes / on 都算开