数据连接 (Data Connection)

uv add "heavenbase[sql]"

import sqlite3


with sqlite3.connect("warehouse.db") as conn:
    conn.execute(
        "CREATE TABLE IF NOT EXISTS customers "
        "(customer_id INTEGER PRIMARY KEY, name TEXT NOT NULL)"
    )
    conn.execute(
        "INSERT OR IGNORE INTO customers(customer_id, name) VALUES (?, ?)",
        (1, "Ada"),
    )

import heavenbase as hb


ws = hb.HeavenBase("warehouse-catalog", preset="debug")
ws.enable_extension("database")

result = ws.database.ingest("warehouse.db", name="warehouse")
database_id = result["database"][0]

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

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