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 に対するファントムフィールドのチェック）でのみ必要です。

---

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 DB です。

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 の基本要素」で定義）を使って DB を解決します。これは Storage の token-verify エンドポイントにクエリし、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} と大文字小文字を区別せずに照合します。SQL には samples の実際のリテラル値を使用します（大文字小文字はそのまま保持します）。一致する値がない場合は、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 に孤立した FK が作成されます
• bounds：上限が無制限の場合は max を省略します（下流では NULL として保存し、0 は使用しません）

---

検証ルール

モデルを確認する際は、次の項目をエラーとして報告します。

• PHANTOM FIELD — 実際の Snowflake テーブルのカラムに存在しないフィールド名
• DANGLING REL — relationship の from／to に指定された tableId がどの dataset にも存在しない
• REL PHANTOM — relationship の on 句で指定されたカラムが、そのテーブルに存在しない
• DANGLING METRIC — metric の dataset に指定された tableId がどの dataset にも存在しない
• METRIC PHANTOM — metric の sql で参照されているカラムが、そのテーブルに存在しない
• AGG ON STRING — STRING カラムに対して SUM／AVG を直接使用している
• SUM ON PCT — カラム名に PCT を含む場合に SUM(...) を使用している
• DUPLICATES — datasets／metrics／relationships 内で name が重複している、または glossary 内で term が重複している
• CONSTRAINT ORPHAN — constraint の metrics[] エントリに一致する semantic-metric 名がない

警告（ブロックしません）：

• Constraint 名に _critical/_warning/_healthy/_review の接尾辞がない

---

CRUD 操作

ユーザーが会話形式でモデルエンティティの追加、編集、削除を依頼した場合は、これらの手順を使用します。 最初に必ず TOKEN、METASTORE、MODEL_UUID を解決してください（上記の Auth & Setup を参照）。

エンティティの追加

上記の形式を使ってペイロードを作成します。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]}")

Constraint を追加する前に — metrics[] の各名前が既存の semantic-metric であることを確認してください。そうしないと、下流の DIM_METRIC_THRESHOLD テーブルに孤立した FK が作成されます。

エンティティの編集

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 にはこうした問題はありません。

⚠ Dataset／relationship の名前変更は連鎖的に反映されません。dataset の semantic name の変更は安全です。tableId を変更すると、それを参照するすべての metrics と relationships が壊れるため、これらの変更は手動で調整してください。

エンティティの削除

削除する前に、必ずユーザーの確認を得てください。metrics の場合は、先に constraint の参照を確認してください。

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')}")

---

運用上の注意点

metric の名前を変更すると CODE_METRIC も変わります — 下流のパイプラインでは、CODE_METRIC = re.sub(r"[^A-Z0-9]+","_", name.upper()).strip("_") によって CODE_METRIC が生成されます。metric の名前を変更すると、CODE_METRIC で結合している SQL が警告なく動作しなくなります。追加による変更を優先してください。

Constraint の重要度は API で 3 レベルのみです — error／warning／info だけでは、4 段階のヘルス UI に対応できません。代わりに、constraint 名の接尾辞に実際の重要度を含めてください。

sql_dialect は snake_case で、値は固定です — 大文字・小文字を含め、正確に 'Snowflake' または 'BigQuery' のいずれかです。camelCase の sqlDialect は 422 missing property 'sql_dialect' で拒否され、小文字の値は 422 value must be one of 'Snowflake', 'BigQuery' で拒否されます。どちらのエラーも一般的な「Validation failed」としてしか表示されないため、原因を誤認しやすくなります。Snowflake と決めつけず、スタックからプロジェクトの実際のバックエンドを確認してください。

modelUUID はプロジェクトごとに異なります — 同じ論理モデルでも、dev と prod では 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 に失敗すると、レイヤーからオブジェクトが欠落した状態になる可能性があります。