数据连接

rtk bash scripts/sync-env.bash -- --extra sql
rtk uv run python demos/data/seed_warehouse.py   # only if warehouse.db is missing

import heavenbase as hb
import heavenbase.ext as hb_ext

ws = hb.HeavenBase("ent-01-warehouse", backends={"main": {"type": "inmem"}})
ws.enable_extension("database")

from heavenbase.utils.files import pj, get_file_dir

DEMOS = get_file_dir("demos", abs=True)
WAREHOUSE_DB = pj(DEMOS, "data", "warehouse.db", abs=True)

result = hb_ext.ingest_database(ws, WAREHOUSE_DB, name="warehouse")
database_id = result["database"][0]

db_row = ws.get(database_id, entity="db-database")
print(db_row["name"], db_row["dialect"])
print(db_row["dsn_summary"])  # credentials redacted

rtk uv run python demos/enterprise/01_connect_and_ingest.py

tables = (
    ws.query("db-table")
    .where({"database_id": database_id})
    .select("table_name", "row_estimate")
    .execute()
)
print(tables.to_str())

customers = next(row for row in tables.rows() if row["table_name"] == "customers")
columns = (
    ws.query("db-column")
    .where({"table_id": customers["object_id"]})
    .select("column_name", "dtype")
    .execute()
)
print(columns.to_str())

discovered = ws.query("db-table").where({"table_name": {"$match": "customer"}}).execute()
print([row["table_name"] for row in discovered.rows()])