目录 (Catalog)

import heavenbase as hb


class Product(hb.Entity):
    identifier = "catalog-product"

    name = hb.field(hb.ShortText).desc("Display name")
    desc = hb.field(hb.MediumText).desc("Catalog description source")
    tags = hb.field(hb.Json).default([]).desc("Catalog tags source")
    price = hb.field(hb.Float)


ws = hb.HeavenBase("core-catalog", backends={"main": {"type": "inmem"}})
ws.register(Product)
ws.upsert_many(
    Product,
    [
        {"object_id": "p1", "name": "Desk", "desc": "Oak writing desk", "tags": ["office", "furniture"], "price": 120.0},
        {"object_id": "p2", "name": "Lamp", "desc": "Adjustable task lamp", "tags": ["office", "lighting"], "price": 35.0},
    ],
)

catalog_rows = (
    ws.query(hb.Catalog)
    .where(hb.Catalog.target_entity == "catalog-product")
    .select("target_id", "target_entity", "name", "desc", "tags")
    .execute()
    .rows()
)

print(catalog_rows[0]["name"])

catalog_row = catalog_rows[0]
product = ws.get(catalog_row["target_id"], entity=catalog_row["target_entity"])

print(product["object_id"])
print(product["price"])

fields = (
    ws.query(hb.MetaSchema)
    .where(hb.MetaSchema.kind == "field")
    .where(hb.MetaSchema.subject_id == "catalog-product")
    .select("field", "dtype", "desc")
    .execute()
    .rows()
)

storage = (
    ws.query(hb.MetaSchema)
    .where(hb.MetaSchema.kind == "storage")
    .where(hb.MetaSchema.subject_id == "catalog-product")
    .select("field", "backend", "strategy")
    .execute()
    .rows()
)

print(fields[0]["field"])
print(storage[0]["backend"])

logical_type_ids = [option.identifier for option in hb.capabilities.logical_types()]
vector_backends = [option.type for option in hb.capabilities.backends(hb.Vector, op="near")]

print("short-text" in logical_type_ids)
print("inmem" in vector_backends)

audit = ws.audit()["catalog"]
print(audit["ok"])

repair_plan = ws.repair(dry_run=True)["catalog"]
print(repair_plan["dry_run"])