扩展系统

import heavenbase as hb

ws = hb.HeavenBase("ext-demo", preset="debug")

print("system" in ws.extensions())
print("prompt" in ws.extensions())
print("sys-capsule" in ws.entities)

print(list(ws.extensions()))

ext_rows = (
    ws.query(hb.MetaSchema)
    .where(hb.MetaSchema.kind == "extension")
    .select("subject_id", "name", "desc")
    .execute()
    .rows()
)

print([row["subject_id"] for row in ext_rows])

import heavenbase as hb
from heavenbase.ext import Extension, ExtensionSpec, register_extension


class AuditLog(hb.Entity):
    identifier = "audit-log"

    event = hb.field(hb.ShortText)
    payload = hb.field(hb.Json).default({})


register_extension(
    Extension(
        ExtensionSpec(
            identifier="audit",
            name="Audit Log",
            desc="Optional audit rows for demo extensions.",
            entities=(AuditLog,),
            tags=("audit",),
        )
    )
)

ws = hb.HeavenBase("ext-custom", preset="debug")
ws.enable_extension("audit")
ws.register(AuditLog)
ws.upsert(
    AuditLog,
    {"object_id": "a1", "name": "startup", "event": "startup", "payload": {"ok": True}},
)

from heavenbase.ext import registered_backend_identifiers, registered_ops

print("sqlite" in registered_backend_identifiers())
print("eq" in registered_ops())

import heavenbase as hb

hb.capabilities.logical_types()
hb.capabilities.backends(hb.Vector, hb.VectorIndex, op="near")