目录 (Catalog)

import heavenbase as hb


class Product(hb.Entity):
    name = hb.field(hb.ShortText)
    desc = hb.field(hb.MediumText)
    tags = hb.field(hb.Json).default([])
    price = hb.field(hb.Float)


ws = hb.HeavenBase("catalog-demo", preset="debug")
ws.register(Product)
ws.upsert(
    Product,
    {
        "object_id": "desk",
        "name": "Oak desk",
        "desc": "A writing desk for very serious snacks.",
        "tags": ["office", "furniture"],
        "price": 120.0,
    },
)

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

hit = rows[0]
product = ws.get(hit["target_id"], entity=hit["target_entity"])
print(product["price"])

rows = ws.query(hb.Catalog).available(stale=True).execute().rows()

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

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

audit = ws.audit()["catalog"]
plan = ws.repair(dry_run=True)["catalog"]

print(audit["ok"], plan["dry_run"])