BlockDB-py API 完整手册

# 数据层主入口
from blockdb import (
    Table,
    BlockTable, Block, Aligned,
    TimeTable,
    Column, filter,
    LogicalType,
    operator, Operator,
)

# 订阅类（L1 / L2 同名，不在顶层暴露，从子包 import）
from blockdb.table import Subscribe, RowEvent          # L1 行事件订阅
from blockdb.block_table import Subscribe              # L2 区块订阅

# 时间桶对齐工具
from blockdb.time_table import align_time_at

# Operator 模块级入口（链式 pipeline）
from blockdb.operator import filter as op_filter, deduplicate as op_dedup

# Meta（建表 / 改 schema / 列表 / 删表）走 dispatcher
from blockdb._grpc_client import blockdb as c

from blockdb import Table

t = Table("demo.event_log")

from blockdb import Table

t = Table("chaintable.event.eth")
rows = t.filter_rows(
    filters="block_height >= 1000 AND name IN ('Transfer', 'Deposit')",
    order_by="block_height asc", limit=200,
)

for row in t.scan_iter("SELECT * FROM demo.event_log"):
    ...   # 处理单行；不要 list() 强转，那会退化成 scan

from blockdb import BlockTable, Block, Aligned
from blockdb.block_table import Subscribe

from blockdb.block_table import Subscribe

from blockdb import Aligned

from blockdb import TimeTable
from blockdb.time_table import align_time_at

from blockdb import Column

from blockdb import filter

from blockdb import Table, Column, filter

t = Table("demo.event_log")
where = filter(
    (Column("created_at") >= 1700000000)
    & ((Column("name") == "login") | (Column("name") == "signup"))
    & ~(Column("value") == 0)
)
t.filter_rows(filters=where, order_by="created_at asc", limit=100)

from blockdb import Operator, operator
from blockdb.operator import filter as op_filter, deduplicate as op_dedup

Operator(source: str = "source_table")

{
  "relations": [
    {"read": {
        "common": {"direct": {}},
        "base_schema": {"names": ["..."]},
        "named_table": {"names": ["source_table"]}
    }},
    {"filter": {
        "common": {"direct": {}},
        "input": {"relation_id": 0},
        "condition": "<scalar_function tree>"
    }},
    {"deduplicate": {
        "common": {"direct": {}},
        "input": {"relation_id": 1},
        "keys": ["<selection>"]
    }}
  ]
}

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

op = (
    op_filter(
        (Column("value") > 0)
        & ((Column("name") == "Transfer") | (Column("name") == "Deposit"))
    )
    .deduplicate(Column("tx_id"), Column("from_addr"))
)
print(op.to_json())

from blockdb.table import Subscribe, RowEvent

from blockdb import LogicalType
from blockdb.logical_type import restore_value

from blockdb._grpc_client import blockdb as c

from blockdb._grpc_client import blockdb as c

# 1) 建 event 类型 BlockTable
c.create_block_table("token.token_transfer.eth", type="event", chain_id="eth")

# 2) 覆盖式写 schema（含 server 自动建的列也必须一起传）
c.save_block_table("token.token_transfer.eth", columns=[
    {"name": "id",              "type": "HASH"},
    {"name": "block_height",    "type": "INTEGER"},
    {"name": "block_timestamp", "type": "TIMESTAMP"},
    {"name": "block_id",        "type": "TX_HASH"},
    {"name": "token_id",        "type": "ADDRESS"},
    {"name": "from_addr",       "type": "ADDRESS"},
    {"name": "to_addr",         "type": "ADDRESS"},
    {"name": "value",           "type": "FLOAT"},
    {"name": "tx_id",           "type": "TXID"},
])

# 3) 建 TimeTable 并把主键改为 ADDRESS
c.create_time_table("time_table_demo")
c.save_time_table("time_table_demo", id_type="ADDRESS")
print(c.list_time_tables(regex="time_table_.*"))