semantic-layer

语义层 — 参考

使用此插件中的命令执行结构化操作： /sl-show · /sl-validate · /sl-build

对于添加／编辑／移除操作，请通过对话完成 — 此技能提供了以下完整的 CRUD 流程。无需斜杠命令。

---

身份验证与设置

按以下回退链解析 TOKEN、STACK 和 METASTORE — 在第一个可用的选项处停止：

1. 环境变量（无需 kbagent）：

import re, os
token     = os.environ.get('KBC_TOKEN', '')
stack_url = os.environ.get('KBC_STACK_URL') or os.environ.get('KBC_URL', '')
if token and stack_url:
    m      = re.search(r'connection\.([\w-]+)\.(gcp|aws|azure)\.keboola\.com', stack_url)
    region = m.group(1) if m else 'us-east4'
    cloud  = m.group(2) if m else 'gcp'
    TOKEN     = token
    STACK     = f'https://connection.{region}.{cloud}.keboola.com'
    METASTORE = f'https://metastore.{region}.{cloud}.keboola.com'

2. kbagent 配置文件（如果已安装 kbagent）：

import json, re, os
cfg_path = os.path.expanduser('~/Library/Application Support/keboola-agent-cli/config.json')
if os.path.exists(cfg_path):
    cfg = json.load(open(cfg_path))
    for alias in sorted(cfg['projects']): print(alias)  # list for user to pick
    p      = cfg['projects'][PROJECT]   # PROJECT = alias chosen by user from list above
    m      = re.search(r'connection\.([\w-]+)\.(gcp|aws|azure)\.keboola\.com', p['stack_url'])
    region = m.group(1) if m else 'us-east4'
    cloud  = m.group(2) if m else 'gcp'
    TOKEN     = p['token']
    STACK     = f'https://connection.{region}.{cloud}.keboola.com'
    METASTORE = f'https://metastore.{region}.{cloud}.keboola.com'

3. 询问用户（以上两种方式均不可用时的回退选项）： 询问以下信息：

• Storage API 令牌 — Keboola UI → Settings → API Tokens
• 连接 URL — 例如：connection.europe-west3.gcp.keboola.com

根据连接 URL 中的区域和云服务推导 STACK 和 METASTORE。

解析完成后：H = {'X-StorageAPI-Token': TOKEN, 'Content-Type': 'application/json'}

CRUD 操作或 /sl-show／/sl-validate 不需要 kbagent。 只有 /sl-build（架构 + SQL 发现）和 /sl-validate --deep （针对 Snowflake 的虚假字段检查）需要 kbagent。

---

API 基本操作

import urllib.request, json

def api_get(path):
    req = urllib.request.Request(f"{METASTORE}{path}", headers={'X-StorageAPI-Token': TOKEN})
    return json.loads(urllib.request.urlopen(req, timeout=15).read()).get('data', [])

def api_post(path, body):
    req = urllib.request.Request(
        f"{METASTORE}{path}", json.dumps(body).encode(), H, method='POST')
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.loads(r.read())

def api_patch(path, body):
    req = urllib.request.Request(
        f"{METASTORE}{path}", json.dumps(body).encode(), H, method='PATCH')
    with urllib.request.urlopen(req, timeout=30) as r:
        return json.loads(r.read())

def api_delete(path):
    req = urllib.request.Request(f"{METASTORE}{path}", headers=H, method='DELETE')
    urllib.request.urlopen(req, timeout=15)

def db_name():
    """Resolve Snowflake DB for the current project: KEBOOLA_<projectId>.
    Caches to /tmp/sl_db_name.txt for the run. Falls back to 'KEBOOLA' on failure."""
    import sys
    cache = '/tmp/sl_db_name.txt'
    if os.path.exists(cache):
        return open(cache).read().strip()
    try:
        req = urllib.request.Request(f"{STACK}/v2/storage/tokens/verify",
                                      headers={'X-StorageApi-Token': TOKEN})
        pid = json.loads(urllib.request.urlopen(req, timeout=15).read())['owner']['id']
        name = f'KEBOOLA_{pid}'
    except Exception as e:
        print(f"⚠ db_name resolve failed ({e}); falling back to KEBOOLA", file=sys.stderr)
        name = 'KEBOOLA'
    open(cache, 'w').write(name)
    return name

端点：

GET    /api/v1/repository/{type}       → {"data": [...]}
POST   /api/v1/repository/{type}       → {"data": {item}}
PATCH  /api/v1/repository/{type}/{id}  → {"data": {item}}   # in-place update
DELETE /api/v1/repository/{type}/{id}

类型： semantic-model · semantic-dataset · semantic-metric · semantic-relationship · semantic-glossary · semantic-constraint

按模型筛选： 对返回的列表使用 i.get('attributes', {}).get('modelUUID') == UUID — ?modelId 查询参数不可靠。

POST 信封（除 semantic-model 外的所有类型）：

{
  "name": "<item name or term>",
  "data": { ...item fields..., "modelUUID": "<UUID>" },
  "branch": "main",
  "schemaVersion": "1.0.0",
  "scope": "project"
}

semantic-model 的 POST 信封：

{
  "name": "<model name>",
  "data": { "name": "<model name>", "description": "...", "sql_dialect": "Snowflake" },
  "branch": "main",
  "schemaVersion": "1.0.0",
  "scope": "project"
}

---

负载结构

semantic-dataset

{
  "name": "fact_revenue",
  "tableId": "out.c-gold.FACT_REVENUE",
  "fqn": "\"KEBOOLA_293\".\"out.c-gold\".\"FACT_REVENUE\"",
  "description": "...",
  "grain": "one row per transaction",
  "primaryKey": ["PK_REVENUE"],
  "fields": [
    { "name": "PK_REVENUE", "type": "string",  "role": "key",       "description": "..." },
    { "name": "AMOUNT",     "type": "decimal",  "role": "measure",   "description": "..." },
    { "name": "PERIOD",     "type": "string",   "role": "dimension", "description": "..." },
    { "name": "INS_DT",     "type": "datetime", "role": "timestamp", "description": "..." }
  ]
}

FQN — 仅按最后一个点拆分 tableId；第一段是项目专属的 Snowflake 数据库：

def fqn(tid, db):
    t = tid.split('.')
    return f'"{db}"."{".".join(t[:-1])}"."{t[-1]}"'
# fqn("out.c-gold.FACT_REVENUE", db_name())
# → "KEBOOLA_293"."out.c-gold"."FACT_REVENUE"

每次运行时，通过 db_name()（定义于上文的 API 基本操作中）解析一次数据库名称 — 此函数会查询存储令牌验证端点，并将 KEBOOLA_<projectId> 缓存到 /tmp/sl_db_name.txt。 切勿硬编码 KEBOOLA — 真实项目使用 KEBOOLA_<projectId>（例如 KEBOOLA_293），单独使用 KEBOOLA 会导致 Snowflake 查询失败。

字段角色：

• PK_*/FK_* → key
• *_DATE / DATE_* / INS_DT / UPD_DT → timestamp
• 数值金额／数值／比率 → measure
• 其他所有字段 → dimension

字段类型： 使用 column_details[].type 中的实际 Snowflake 类型。 STRING→string · NUMERIC→decimal 或 integer · 切勿将 STRING 覆盖为 boolean/date。

semantic-metric

{
  "name": "Total Revenue",
  "sql": "SUM(\"FACT_REVENUE\".\"AMOUNT\")",
  "dataset": "out.c-gold.FACT_REVENUE",
  "description": "..."
}

• dataset 字段应为 tableId，而不是数据集名称
• 仅对真实列使用 SUM／AVG／COUNT(DISTINCT)
• 切勿对 _PCT／比率列使用 SUM — 应使用 AVG
• VERSION 表：只有在探测列的实际不同值之后，才生成 SUM(CASE WHEN "T"."<col>" = '<value>' THEN ...) 指标。 使用 /sl-build 第 2.5 步中的 kbagent query_data MCP 工具（会写入 /tmp/sl_version_samples.json）— 临时使用时也可：

  import subprocess, json, csv, io
  payload = json.dumps({'query_name': f'probe {COL}',
      'sql_query': f'SELECT DISTINCT "{COL}" AS V FROM "{SCHEMA}"."{TABLE}" LIMIT 20'})
  r = subprocess.run(['kbagent','--json','tool','call','query_data',
                      '--project', PROJECT, '--input', payload],
                     capture_output=True, text=True)
  d = json.loads(r.stdout)['data']
  samples = set()
  for res in d.get('results', []):
      if res.get('isError'): continue
      for piece in res.get('content', []):
          p = json.loads(piece) if isinstance(piece, str) else piece
          csv_text = p.get('csv_data', '') if isinstance(p, dict) else ''
          for row in csv.DictReader(io.StringIO(csv_text)):
              if row.get('V'): samples.add(row['V'])
  

  仅当 samples 包含已识别的字面值时，才应用 VERSION 规则 — 不区分大小写地匹配 {actual, budget, plan, forecast, baseline, target}。 将 samples 中的实际字面值（保留大小写）代入 SQL。如果没有匹配项，则跳过 VERSION 条件指标，并在模型描述中注明："未为 <col> 生成 VERSION 风格的细分 — 不同值为 <samples>。"

semantic-relationship

{
  "name": "fact_revenue_to_time",
  "from": "out.c-gold.FACT_REVENUE",
  "to":   "out.c-gold.DIM_DATE",
  "on":   "\"FACT_REVENUE\".\"PERIOD\" = \"DIM_DATE\".\"PK_DATE\"",
  "type": "left"
}

• from／to 是 tableId · on 使用不带限定的表名（tableId 的最后一段）
• type：left 或 inner

semantic-glossary

{ "term": "EBITDA", "definition": "Earnings before interest, taxes, depreciation and amortization." }

semantic-constraint

{
  "name": "net_margin_critical",
  "constraintType": "range",
  "metrics": ["Net Profit Margin"],
  "ruleExpression": { "bounds": { "min": -2.0, "max": 0.05 } },
  "severity": "error"
}

• API 的 severity 仅接受：error／warning／info
• 在名称后缀中编码 4 级健康状态：_critical／_warning／_healthy／_review
• 下游流水线会解析该后缀；severity 是次要字段
• metrics[] 必须包含完全匹配的指标名称 — 任何不匹配都会在 DIM_METRIC_THRESHOLD 中产生孤立外键
• bounds：对于无上限的上界，省略 max（下游存储为 NULL，绝不能为 0）

---

验证规则

检查模型时，将以下情况标记为错误：

• PHANTOM FIELD — 字段名不存在于 Snowflake 表的实际列中
• DANGLING REL — 关系中的 from/to tableId 不在任何数据集中
• REL PHANTOM — 关系 on 子句中的列不存在于其所属表中
• DANGLING METRIC — 指标的 dataset tableId 不在任何数据集中
• METRIC PHANTOM — 指标 sql 中引用的列不存在于其所属表中
• AGG ON STRING — 对 STRING 列直接使用 SUM/AVG
• SUM ON PCT — 对列名包含 PCT 的列使用 SUM(...)
• DUPLICATES — 数据集/指标/关系中存在重复的 name，或词汇表中存在重复的 term
• CONSTRAINT ORPHAN — 约束的 metrics[] 条目没有匹配的语义指标名称

警告（不阻止操作）：

• 约束名称缺少 _critical/_warning/_healthy/_review 后缀

---

CRUD 操作

当用户通过对话要求添加、编辑或移除模型实体时，请使用这些流程。 始终先解析 TOKEN、METASTORE 和 MODEL_UUID（见上文“身份验证与设置”）。

添加实体

使用上述结构构建载荷。POST 前先向用户展示载荷。 所有 CRUD 代码块均假设已定义 API Primitives 中的 api_get/api_post/api_delete。

import urllib.error

TYPE = 'semantic-metric'   # replace with actual type
ITEM = { }                 # replace with actual payload

body = {
    "name": ITEM.get('name') or ITEM.get('term'),
    "data": {**ITEM, "modelUUID": MODEL_UUID},
    "branch": "main", "schemaVersion": "1.0.0", "scope": "project"
}
try:
    r = api_post(f"/api/v1/repository/{TYPE}", body)
    print(f"✓ Created {r['data']['id']}")
except urllib.error.HTTPError as e:
    print(f"✗ {e.code}: {e.read().decode()[:300]}")

添加约束前 — 验证 metrics[] 中的每个名称都是现有语义指标，否则该约束会在下游 DIM_METRIC_THRESHOLD 表中创建孤立外键。

编辑实体

使用 PATCH 原地编辑。 只发送发生变化的字段——对象会保留其 UUID 并生成新修订，因此历史记录得以保留，任何通过 UUID 引用它的内容仍然有效。 继续操作前，始终向用户展示差异并征得确认。

import urllib.error, re

# 1. Fetch and find item
TYPE = 'semantic-metric'   # replace with actual type
all_items = api_get(f"/api/v1/repository/{TYPE}")
items  = [i for i in all_items if i.get('attributes', {}).get('modelUUID') == MODEL_UUID]
target = next((i for i in items
               if i['attributes'].get('name','').lower() == TARGET_NAME.lower()), None)
if not target:
    print("Not found. Available:", [i['attributes'].get('name') for i in items])

# 2. Decide the change (a partial patch — not the whole object)
OLD_NAME = target['attributes'].get('name', '')
CHANGES  = {}                       # e.g. {'sql': '<new sql>'} or {'name': 'Total Revenue'}
NEW_NAME = CHANGES.get('name', OLD_NAME)

# 3. If renaming a metric — find constraints to cascade-update
is_rename = TYPE == 'semantic-metric' and NEW_NAME != OLD_NAME
affected_constraints = []
if is_rename:
    all_c = api_get("/api/v1/repository/semantic-constraint")
    affected_constraints = [
        c for c in all_c
        if c.get('attributes', {}).get('modelUUID') == MODEL_UUID
        and OLD_NAME in (c.get('attributes', {}).get('metrics') or [])
    ]
    old_code = re.sub(r"[^A-Z0-9]+", "_", OLD_NAME.upper())
    new_code = re.sub(r"[^A-Z0-9]+", "_", NEW_NAME.upper())
    print(f"CODE_METRIC: {old_code} → {new_code}  ⚠ update any pipeline SQL joining on this key")
    if affected_constraints:
        print(f"Constraints to auto-update: {[c['attributes']['name'] for c in affected_constraints]}")

# 4. PATCH in place. Include `name` at the envelope top level only when it changed,
#    so the metastore's own `meta.name` stays in sync with the payload.
body = {"data": CHANGES}
if is_rename or 'term' in CHANGES:
    body["name"] = CHANGES.get('name') or CHANGES.get('term')
try:
    r = api_patch(f"/api/v1/repository/{TYPE}/{target['id']}", body)
    print(f"✓ Updated {r['data']['id']} (revision {r['data'].get('meta', {}).get('revision')})")
except urllib.error.HTTPError as e:
    # Nothing was deleted, so there is nothing to roll back — the object is untouched.
    print(f"✗ PATCH failed ({e.code}): {e.read().decode()[:300]}")
    raise

# 5. Cascade constraint updates on rename — also in place
for c in affected_constraints:
    metrics = [NEW_NAME if m == OLD_NAME else m for m in (c['attributes'].get('metrics') or [])]
    try:
        api_patch(f"/api/v1/repository/semantic-constraint/{c['id']}", {"data": {"metrics": metrics}})
        print(f"  ✓ Constraint updated: {c['attributes']['name']}")
    except urllib.error.HTTPError as e:
        print(f"  ✗ {c['attributes']['name']}: {e.code}")

不要通过 DELETE + POST 编辑。 这样会销毁对象的 UUID 和修订历史，破坏任何通过 UUID 引用它的内容；如果 POST 失败，还会导致该层在一段时间内缺少对象。PATCH 不会带来这些问题。

⚠ 数据集/关系重命名不会级联更新。重命名数据集的语义名称是安全的。更改其 tableId 会破坏所有指向它的指标和关系——请手动协调这些更改。

移除实体

删除前始终向用户确认。对于指标，请先检查约束引用。

TYPE = 'semantic-metric'   # replace with actual type
# (find target same as Edit step 1 above)

# Check constraint references before deleting a metric
if TYPE == 'semantic-metric':
    all_c = api_get("/api/v1/repository/semantic-constraint")
    refs = [c['attributes']['name'] for c in all_c
            if target['attributes']['name'] in (c.get('attributes', {}).get('metrics') or [])]
    if refs:
        print(f"⚠ Constraints referencing this metric: {refs}")
        print("Deleting will create orphan entries in downstream DIM_METRIC_THRESHOLD.")
        # Ask user to confirm before continuing

# Delete
api_delete(f"/api/v1/repository/{TYPE}/{target['id']}")
print(f"✓ Deleted: {target['attributes'].get('name')}")

---

操作注意事项

重命名指标会改变其 CODE_METRIC — 下游流水线使用 re.sub(r"[^A-Z0-9]+","_", name.upper()).strip("_") 推导 CODE_METRIC。如果重命名指标，任何通过 CODE_METRIC 进行连接的 SQL 都会静默失效。优先采用增量式更改。

约束严重性只有 3 个 API 级别 — 对于包含 4 个等级的健康度 UI，error/warning/info 并不足够。请在约束名称后缀中标明实际严重性。

sql_dialect 使用 snake_case，且取值固定 — 只能是准确拼写并区分大小写的 'Snowflake' 或 'BigQuery'。camelCase 的 sqlDialect 会被拒绝，并返回 422 missing property 'sql_dialect'；小写值会被拒绝，并返回 422 value must be one of 'Snowflake', 'BigQuery'。这两种错误都只会显示为通用的“验证失败”，因此很容易误诊。请从技术栈中确认项目实际使用的后端，不要想当然地认为是 Snowflake。

每个项目的 modelUUID 都不同 — 同一个逻辑模型在开发环境和生产环境中有不同的 UUID。发布时，先获取目标项目的模型列表以查找其 UUID，然后在 POST 每个项目之前替换其 modelUUID。

执行破坏性更改前创建快照：

for t in ['semantic-metric','semantic-dataset','semantic-glossary',
          'semantic-constraint','semantic-relationship']:
    json.dump(api_get(f'/api/v1/repository/{t}'),
              open(f'/tmp/sl_backup_{t}.json','w'), indent=2)

使用 PATCH 编辑，绝不要使用 DELETE + POST — PATCH /api/v1/repository/{type}/{id} 会原地更新对象，保留其 UUID 并递增修订号。删除后重新 POST 会生成新的 UUID、重置修订历史、破坏任何引用旧 UUID 的内容；如果 POST 失败，还可能导致该层缺少对象。