数据目录阶段 4A 影响面:新增接口列出引用某张表的已发布和受管模板,区分直接引用和可能涉及、合并查询按输入追溯,表详情和保存纠正后都显示

This commit is contained in:
ThearchyHelios
2026-10-03 11:52:49 +08:00
parent 1c806fa3a0
commit 958e096e56
11 changed files with 741 additions and 6 deletions
+14 -1
View File
@@ -6,6 +6,7 @@
- POST /catalog/{table}/review 单项审阅:确认、驳回、撤销审阅
- POST /catalog/{table}/patch/preview 目录修改提案的预览(只读):对着当前目录算改前、改后
- POST /catalog/{table}/patch 保存目录修改提案(乐观锁),记为人工确认
- GET /catalog/{table}/impact 影响面:引用这张表的已发布、受管模板(data/catalog_impact.py)
- POST /catalog/draft 同步起草(注释、外键、命名推断,可选模型)
规则都在 data/catalog.py,这里只做取源、转换形状和把异常翻成状态码:版本不符 409,结构不合规、审阅路径
@@ -22,7 +23,7 @@ from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.runs import actor_of
from app.data import catalog, introspect, table_versions
from app.data import catalog, catalog_impact, introspect, table_versions
from app.db.base import get_session
from app.db.models import DataSource
@@ -280,3 +281,15 @@ async def save_catalog_patch(source_id: str, table: str, payload: CatalogPatchIn
except catalog.CatalogInvalid as e:
raise _invalid(e) from e
return await _detail(session, source, key, entry)
@router.get("/{source_id}/catalog/{table}/impact")
async def catalog_table_impact(source_id: str, table: str,
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
"""引用这张表的已发布、受管模板,每个模板按当前的已发布版本算:调用工具节点的 SQL 用到了这张表算直接引用,
Agent 绑定了这个源的查询工具算可能涉及,合并查询按输入追溯。改目录之前、保存纠正之后都看它。"""
row, source = await _resolved(session, source_id)
key, _ = await _table_key(session, source, table)
tables = _tables_of(source)
templates = await catalog_impact.table_impact(session, row.name, key, list(tables) if tables else None)
return {"table": key, "templates": templates}
+289
View File
@@ -0,0 +1,289 @@
"""数据目录的影响面与发布时的目录版本(阶段 4A:维护闭环)。
目录改了一项(「在园人数是存量」),用到这张表的模板算出来的数可能就该变了。这里回答两件事:
- **影响面**:哪些已发布、受管的模板用到了这张表(table_impact)。按每个模板当前的已发布版本算,不看画布上的
草稿——正式运行跑的是已发布的那一版。调用工具节点写死的 SQL 用到了这张表算「直接引用」;Agent 绑定了这个源
的查询工具算「可能涉及」(SQL 是运行时写的,静态看不出);合并查询按它的输入一路追溯。
- **发布时的目录版本**:发布时记下模板 SQL 用到的每张表的目录版本 {源名: {表名: 版本}}(catalog_versions_for),
从这一版发起正式运行时和当前的比(catalog_drift),有变化只提醒、不拦运行——目录是给人和助手看的说明,
不是运行的输入;拦下来等于让一次文字修订卡住正式出具。
表名的取法和证据台账同一个口径:evidence.sql_tables 取 FROM / JOIN 后面的表(去掉注释、字符串、CTE),
去掉 schema 前缀后按 name_key 对到表结构上(和 catalog.table_usage 一致)。
"""
from __future__ import annotations
import logging
from collections.abc import Iterable, Mapping
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.data.names import name_key
logger = logging.getLogger(__name__)
#: 影响程度:direct 直接引用(SQL 里写着这张表);possible 可能涉及(Agent 运行时自己写 SQL)
IMPACTS = ("direct", "possible")
def _lookup(tables: Iterable[str]) -> dict[str, str]:
out: dict[str, str] = {}
for name in tables:
out.setdefault(name_key(name), name)
return out
def sql_table_keys(sql: str, tables: Iterable[str] | None) -> set[str]:
"""SQL 用到的表,对到表结构的写法上。tables 不给(不知道表结构)时用 SQL 里去掉 schema 前缀的原名。"""
from app.engine.evidence import sql_tables
try:
names = sql_tables(sql or "")
except Exception: # noqa: BLE001 - 解析不了就当没用到表:影响面和版本记录都是提示,不能因此报错
return set()
shorts = [str(n).rsplit(".", 1)[-1] for n in names]
if tables is None:
return set(shorts)
lookup = _lookup(tables)
return {hit for s in shorts if (hit := lookup.get(name_key(s))) is not None}
def graph_sql_tables(spec: Any, tables_by_source: Mapping[str, Iterable[str]] | None = None) -> dict[str, set[str]]:
"""一张图里调用工具节点写死的 SQL 用到的表:{源名: {表名}}。Agent 的 SQL 运行时才有,不在这里。
tables_by_source 是各源表结构里的表名(schema_cache 的 tables 的键):给了的源按它规范表名、对不上的不算;
没给的源用 SQL 里的原名。
"""
from app.data.sqlcheck import tool_query
out: dict[str, set[str]] = {}
for node in getattr(spec, "nodes", []) or []:
found = tool_query(node)
if not found:
continue
source, sql = found
known = (tables_by_source or {}).get(source)
hits = sql_table_keys(sql, list(known) if known is not None else None)
if hits:
out.setdefault(source, set()).update(hits)
return out
def _kind(node: Any) -> str:
return str(getattr(node.type, "value", node.type))
def _setting(spec: Any, node: Any, key: str) -> Any:
"""节点配置的实际取值:节点上没写就取图级 defaults(和运行时 NodeContext.cfg 同一个规则)。"""
value = (node.config or {}).get(key)
return (getattr(spec, "defaults", None) or {}).get(key) if value in (None, "") else value
def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | None) -> list[dict[str, Any]]:
"""一张图里哪些节点用到了某个源的某张表,按图里的顺序。每项 {node_id, label, type, impact, via?, member?}。
- 调用工具节点:调用 db_query__<源>、SQL 里有这张表 → direct。
- Agent 节点、协作成员:绑定了 db_query__<源> → possible(每个成员各一项,带 member)。
- 合并查询:输入(别名 → 节点)里有用到这张表的 → 取输入里最强的那一级,via 写经由哪些输入。合并的合并
一路追到底;成环(校验会报错的图)时停下。
"""
from app.data.sqlcheck import tool_query
from app.tools.datasource import QUERY_PREFIX
tool = f"{QUERY_PREFIX}{source}"
wanted = name_key(table)
known = list(tables) if tables is not None else None
nodes = list(getattr(spec, "nodes", []) or [])
by_id = {n.id: n for n in nodes}
own: dict[str, dict[str, Any]] = {}
out: list[dict[str, Any]] = []
def base(node: Any, impact: str, **extra: Any) -> dict[str, Any]:
return {"node_id": node.id, "label": node.title, "type": _kind(node), "impact": impact, **extra}
for node in nodes:
kind = _kind(node)
if kind == "tool":
found = tool_query(node)
if found and found[0] == source and wanted in {name_key(t) for t in sql_table_keys(found[1], known)}:
own[node.id] = base(node, "direct")
elif kind == "agent":
if tool in [str(t).strip() for t in _setting(spec, node, "tools") or [] if isinstance(t, str)]:
own[node.id] = base(node, "possible")
elif kind == "supervisor":
for member in _setting(spec, node, "agents") or []:
if isinstance(member, Mapping) and tool in [str(t).strip() for t in member.get("tools") or []
if isinstance(t, str)]:
out.append(base(node, "possible", member=str(member.get("name") or "")))
resolved: dict[str, dict[str, Any] | None] = {}
def through(node_id: str, trail: frozenset[str]) -> dict[str, Any] | None:
"""合并查询节点经由输入用到这张表的样子;用不到返回 None。"""
if node_id in resolved:
return resolved[node_id]
node = by_id.get(node_id)
if node is None or _kind(node) != "merge" or node_id in trail:
return own.get(node_id)
via: list[dict[str, Any]] = []
best: str | None = None
inputs = (node.config or {}).get("inputs")
for alias, target in (inputs.items() if isinstance(inputs, Mapping) else []):
if not isinstance(target, str):
continue
hit = own.get(target) or through(target, trail | {node_id})
if hit is None:
continue
via.append({"node_id": target, "label": by_id[target].title if target in by_id else target, "alias": alias})
best = "direct" if hit["impact"] == "direct" or best == "direct" else "possible"
resolved[node_id] = base(node, best, via=via) if best else None
return resolved[node_id]
for node in nodes:
if node.id in own:
out.append(own[node.id])
elif _kind(node) == "merge" and (hit := through(node.id, frozenset())):
out.append(hit)
order = {n.id: i for i, n in enumerate(nodes)}
out.sort(key=lambda r: order.get(r["node_id"], 0))
return out
async def _published(session: AsyncSession) -> list[tuple[Any, Any]]:
"""每个发布过的模板和它当前的已发布版本:[(Workflow, WorkflowVersion)]。"""
from app.db.models import Workflow, WorkflowVersion
rows = (await session.execute(
select(Workflow, WorkflowVersion)
.join(WorkflowVersion, (WorkflowVersion.workflow_id == Workflow.id)
& (WorkflowVersion.version == Workflow.published_version))
.where(Workflow.published_version.is_not(None))
)).all()
return [(w, v) for w, v in rows]
async def table_impact(session: AsyncSession, source: str, table: str,
tables: Iterable[str] | None) -> list[dict[str, Any]]:
"""引用某个源某张表的已发布、受管模板,直接引用的在前,同级按名字。
每项 {workflow_id, name, version, level, impact, nodes}:version 是当前的已发布版本,level 是它发布时的级别
(published / governed;记级别之前发布的老版本按工作流现在的状态)。图解析不了的版本跳过。
"""
from app.engine.schema import GraphSpec
known = list(tables) if tables is not None else None
out: list[dict[str, Any]] = []
for workflow, version in await _published(session):
try:
spec = GraphSpec.model_validate(version.graph or {})
except Exception: # noqa: BLE001 - 解析不了的版本发起不了正式运行,也就谈不上影响
continue
refs = template_refs(spec, source, table, known)
if not refs:
continue
level = version.level or (workflow.status if workflow.status in ("published", "governed") else "published")
out.append({
"workflow_id": workflow.id, "name": workflow.name, "version": version.version, "level": level,
"impact": "direct" if any(r["impact"] == "direct" for r in refs) else "possible", "nodes": refs,
})
out.sort(key=lambda t: (t["impact"] != "direct", t["name"]))
return out
# ==========================================================================
# 发布时的目录版本
# ==========================================================================
async def _sources_by_name(session: AsyncSession, names: Iterable[str]) -> dict[str, Any]:
"""启用着的数据源:{源名: (DataSource 行, 用来读表结构的源)}。上传源解析成当前快照的视图,解析不了用原行。"""
from app.data import table_versions
from app.db.models import DataSource
wanted = sorted({n for n in names if n})
if not wanted:
return {}
rows = (await session.execute(
select(DataSource).where(DataSource.name.in_(wanted), DataSource.enabled.is_(True))
)).scalars()
out: dict[str, Any] = {}
for row in rows:
try:
view = await table_versions.resolve_source(session, row)
except table_versions.SnapshotMissing:
view = row
out[row.name] = (row, view)
return out
def _schema_tables(view: Any) -> list[str] | None:
tables = (getattr(view, "schema_cache", None) or {}).get("tables")
return list(tables) if isinstance(tables, Mapping) and tables else None
async def catalog_versions_for(session: AsyncSession, spec: Any) -> dict[str, dict[str, int]]:
"""发布时要记下的目录版本:{源名: {表名: 版本}},只收模板 SQL 用到的表;还没有目录的表记 0。
记 0 是有意的:发布之后有人给这张表补了目录,同样算「自发布以来有变化」。找不到的源(停用、删了)不记:
没有目录可比,正式运行时工具自己会报源不存在。读库出错时返回空 dict——记不下版本不该挡住发布。
"""
from app.data import catalog
try:
names = graph_sql_tables(spec).keys()
sources = await _sources_by_name(session, names)
tables_by_source = {name: _schema_tables(view) for name, (_, view) in sources.items()}
used = graph_sql_tables(spec, {k: v for k, v in tables_by_source.items() if v is not None})
out: dict[str, dict[str, int]] = {}
for name, tables in sorted(used.items()):
if name not in sources:
continue
entries = await catalog.read_catalog(session, sources[name][0].id)
out[name] = {t: (entries[t].version if t in entries else 0) for t in sorted(tables)}
return out
except Exception: # noqa: BLE001
logger.exception("记录发布时的目录版本失败,本次不记")
return {}
async def catalog_drift(session: AsyncSession, recorded: Mapping[str, Any] | None) -> list[dict[str, Any]]:
"""发布时记下的目录版本和现在比:变了的表 [{source, source_id, table, label, published, current}]。
没记过(这个功能之前发布的版本)返回空列表——不知道当时是哪一版,不能说它变了。现在找不到的源跳过(理由同
catalog_versions_for)。label 是表现在的中文名(没有或被驳回时为 None)。读库出错返回空列表:提醒是附加的。
"""
from app.data import catalog
if not isinstance(recorded, Mapping) or not recorded:
return []
try:
sources = await _sources_by_name(session, recorded.keys())
out: list[dict[str, Any]] = []
for name in sorted(recorded):
pinned = recorded[name]
if name not in sources or not isinstance(pinned, Mapping):
continue
row = sources[name][0]
entries = await catalog.read_catalog(session, row.id)
for table in sorted(pinned):
then = pinned[table]
entry = entries.get(table)
now = entry.version if entry else 0
if isinstance(then, int) and then == now:
continue
label = (entry.notes.get("label") if entry else None) or {}
out.append({"source": name, "source_id": row.id, "table": table,
"label": label.get("value") if isinstance(label, dict) and label.get("status") != "rejected"
else None,
"published": then if isinstance(then, int) else None, "current": now})
return out
except Exception: # noqa: BLE001
logger.exception("比对发布时的目录版本失败,本次不提醒")
return []
__all__ = ["IMPACTS", "catalog_drift", "catalog_versions_for", "graph_sql_tables", "sql_table_keys", "table_impact",
"template_refs"]
+185
View File
@@ -0,0 +1,185 @@
"""目录的影响面:改了一张表的目录,哪些已发布、受管的模板会受影响(阶段 4A)。
GET /api/datasources/{id}/catalog/{table}/impact,每个模板按它当前的已发布版本算(不是画布上的草稿):
- 调用工具节点调用这个源的 db_query__,SQL 里用到了这张表:直接引用。表名的取法和证据台账同一个口径
(evidence.sql_tables,按 name_key 对到表结构上)。
- Agent 节点(协作成员也算)绑定了这个源的查询工具:可能涉及——SQL 是运行时写的,静态看不出用不用这张表。
- 合并查询节点按它的输入追溯:输入里有直接引用的,它也算直接引用,并写明经由哪个输入。
- 没发布过的草稿、别的源的同名表、只在未发布草稿里用到的,都不算。
"""
from __future__ import annotations
import uuid
from types import SimpleNamespace
import pytest
from httpx import ASGITransport, AsyncClient
from app.data.engine import engines
from app.data.introspect import introspect
# 模型要在建表之前注册进 metadata:conftest 的建表只导入了 app.db.base,这个文件单独跑时没有别的模块先导入它们
from app.db import models # noqa: F401
_CACHE: dict[str, dict] = {}
async def _schema(path: str) -> dict:
if path not in _CACHE:
probe = SimpleNamespace(id=f"probe-{uuid.uuid4().hex[:6]}", name="scenic", kind="sqlite", database=path,
host=None, port=None, username=None, password=None, options={}, readonly=True,
description="", schema_cache={}, origin="manual")
try:
_CACHE[path] = await introspect(probe)
finally:
await engines.invalidate(probe.id)
return _CACHE[path]
def _node(nid, ntype, label=None, **config):
return {"id": nid, "type": ntype, "data": {"label": label or nid, "config": config}}
def _graph(*nodes):
nodes = [_node("start", "input"), *nodes, _node("out", "output", fields=[])]
return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
def _query(nid, source, sql, label=None):
return _node(nid, "tool", label, tool=f"db_query__{source}", args={"sql": sql})
@pytest.fixture
async def world(scenic_db):
"""一个景区数据源、另一个同结构的源,和几份模板;用完全部删掉。"""
from app.db.base import SessionLocal
from app.db.models import DataSource, Workflow, WorkflowVersion
tag = uuid.uuid4().hex[:8]
src, other = f"scenic_{tag}", f"mirror_{tag}"
created: list[str] = []
async with SessionLocal() as session:
rows = []
for name in (src, other):
row = DataSource(name=name, kind="sqlite", database=scenic_db, readonly=True)
row.schema_cache = await _schema(scenic_db)
session.add(row)
rows.append(row)
await session.commit()
async def template(name, versions, *, published=None, level="published"):
wf = Workflow(name=f"{name}-{tag}", graph=versions[-1], version=len(versions),
status=level if published else "draft", published_version=published)
session.add(wf)
await session.flush()
for i, g in enumerate(versions, start=1):
session.add(WorkflowVersion(workflow_id=wf.id, version=i, graph=g,
level=level if published == i else None))
created.append(wf.id)
return wf
direct = await template("入园日报", [_graph(_query("q", src, "SELECT COUNT(*) FROM main.VISITS v WHERE v.status = 1",
"查询入园人数"))], published=1)
agent = await template("入园分析", [_graph(_node("ask", "agent", "分析入园", prompt="分析",
tools=[f"db_query__{src}", f"db_schema__{src}"]))],
published=1, level="governed")
team = await template("协作分析", [_graph(_node("team", "supervisor", "协作", agents=[
{"name": "analyst", "tools": [f"db_query__{src}"]}]))], published=1)
merged = await template("渠道合并", [_graph(
_query("q1", src, "SELECT id, park_id FROM visits", "入园"),
_query("q2", src, "SELECT id, channel_id FROM orders", "订单"),
_node("m", "merge", "合并入园和订单", inputs={"a": "q1", "b": "q2"}, sql="SELECT * FROM a JOIN b ON a.id = b.id"),
_node("m2", "merge", "再合并", inputs={"x": "m"}, sql="SELECT * FROM x"),
)], published=1)
# 不算的:草稿、别的源的同名表、只在未发布的新版本里用到
await template("草稿", [_graph(_query("q", src, "SELECT * FROM visits"))])
await template("镜像", [_graph(_query("q", other, "SELECT * FROM visits"))], published=1)
await template("后来才用", [_graph(_query("q", src, "SELECT * FROM orders")),
_graph(_query("q", src, "SELECT * FROM visits"))], published=1)
# 已发布的是旧版本:画布上后来删掉了这张表,影响面仍按已发布的那一版算
stale = await template("改过画布", [_graph(_query("q", src, "SELECT * FROM visits")),
_graph(_query("q", src, "SELECT * FROM orders"))], published=1)
await session.commit()
ids = SimpleNamespace(source=rows[0].id, other=rows[1].id, src=src, direct=direct.id, agent=agent.id,
team=team.id, merged=merged.id, stale=stale.id)
yield ids
async with SessionLocal() as session:
for wid in created:
if (wf := await session.get(Workflow, wid)) is not None:
await session.delete(wf)
for sid in (ids.source, ids.other):
if (row := await session.get(DataSource, sid)) is not None:
await session.delete(row)
await session.commit()
async def _impact(source_id: str, table: str) -> dict:
from app.main import app
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
r = await client.get(f"/api/datasources/{source_id}/catalog/{table}/impact")
assert r.status_code == 200, r.text
return r.json()
async def test_impact_classifies_direct_possible_and_merge(world):
out = await _impact(world.source, "visits")
assert out["table"] == "visits"
by_id = {t["workflow_id"]: t for t in out["templates"]}
assert set(by_id) == {world.direct, world.agent, world.team, world.merged, world.stale}
direct = by_id[world.direct]
assert direct["impact"] == "direct" and direct["version"] == 1 and direct["level"] == "published"
assert [(n["node_id"], n["label"], n["impact"]) for n in direct["nodes"]] == [("q", "查询入园人数", "direct")]
agent = by_id[world.agent]
assert agent["impact"] == "possible" and agent["level"] == "governed"
assert [(n["node_id"], n["impact"]) for n in agent["nodes"]] == [("ask", "possible")]
assert by_id[world.team]["impact"] == "possible"
assert by_id[world.team]["nodes"][0]["member"] == "analyst"
merged = by_id[world.merged]
assert merged["impact"] == "direct"
nodes = {n["node_id"]: n for n in merged["nodes"]}
assert set(nodes) == {"q1", "m", "m2"} # 查订单的 q2 不涉及这张表
assert nodes["m"]["impact"] == "direct" and nodes["m"]["via"] == [{"node_id": "q1", "label": "入园", "alias": "a"}]
# 合并的合并:一路追到底
assert nodes["m2"]["impact"] == "direct" and nodes["m2"]["via"] == [{"node_id": "m", "label": "合并入园和订单",
"alias": "x"}]
# 直接引用的排在前面
impacts = [t["impact"] for t in out["templates"]]
assert impacts == sorted(impacts, key=lambda i: i != "direct")
async def test_impact_of_another_table(world):
out = await _impact(world.source, "orders")
by_id = {t["workflow_id"]: t for t in out["templates"]}
# 已发布的旧版本只查了入园记录:画布上后来改成查订单,不算
assert world.stale not in by_id
assert by_id[world.merged]["impact"] == "direct"
assert {n["node_id"] for n in by_id[world.merged]["nodes"]} == {"q2", "m", "m2"}
# 只绑了查询工具的 Agent 对每张表都是「可能涉及」
assert by_id[world.agent]["impact"] == "possible"
assert world.direct not in by_id
async def test_impact_unknown_table_is_404(world):
from app.main import app
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
r = await client.get(f"/api/datasources/{world.source}/catalog/ghost_table/impact")
assert r.status_code == 404
def test_graph_sql_tables_uses_the_evidence_rules():
"""发布时记目录版本和影响面共用这一份:只看调用工具节点写死的 SQL,表名按 name_key 对到表结构上。"""
from app.data.catalog_impact import graph_sql_tables
from app.engine.schema import GraphSpec
spec = GraphSpec.model_validate(_graph(
_query("q1", "shop", "SELECT * FROM main.Orders o JOIN order_items i ON i.order_id = o.id -- FROM ghost"),
_query("q2", "shop", "WITH t AS (SELECT 1) SELECT * FROM t, customers"),
_query("q3", "other", "SELECT * FROM orders"),
_node("ask", "agent", tools=["db_query__shop"]),
))
tables = {"shop": {"orders": {}, "order_items": {}, "customers": {}}, "other": {"orders": {}}}
assert graph_sql_tables(spec, tables) == {"shop": {"orders", "order_items", "customers"}, "other": {"orders"}}
+4 -1
View File
@@ -1,5 +1,5 @@
import type {
ActivateSnapshotBody, ActivateSnapshotOut, AiPreview, CatalogDetail, CatalogDraftOut, CatalogList, CatalogNotes,
ActivateSnapshotBody, ActivateSnapshotOut, AiPreview, CatalogDetail, CatalogDraftOut, CatalogImpact, CatalogList, CatalogNotes,
CatalogPatchPreview, CatalogPatchSubmit, CatalogReviewAction, CommitOut, CurrentRecipe, EditPreview, EditRequest, ImportRecord,
ManifestOut, PurgeRawBody, PurgeRawOut, QuestionAnswer, Recipe, RedraftRulesOut, RemovePeriodBody, RemovePeriodOut,
RevokeAcceptanceBody, RevokeAcceptanceOut, SnapshotOut, Staging,
@@ -669,6 +669,9 @@ export const api = {
/** 单项审阅。path:表级项写字段名(grain);列级项 columns.<列名>.<字段>;关系 relations.<编号> */
review: (sourceId: string, table: string, body: { path: string; action: CatalogReviewAction; if_version: number }) =>
post<CatalogDetail>(`/datasources/${encodeURIComponent(sourceId)}/catalog/${encodeURIComponent(table)}/review`, body),
/** 影响面:引用这张表的已发布、受管模板(直接引用 / 可能涉及,合并查询按输入追溯) */
impact: (sourceId: string, table: string, opts?: RequestOptions) =>
get<CatalogImpact>(`/datasources/${encodeURIComponent(sourceId)}/catalog/${encodeURIComponent(table)}/impact`, opts),
/** 目录修改提案的预览:对着当前目录重算改前、改后,不写库(保存遇到 409 后重新载入用) */
patchPreview: (sourceId: string, table: string, changes: CatalogPatchSubmit[]) =>
post<CatalogPatchPreview>(
+122
View File
@@ -0,0 +1,122 @@
import { useCallback, useEffect, useRef, useState } from 'react'
import { Link } from 'react-router-dom'
import { GitBranch } from 'lucide-react'
import clsx from 'clsx'
import { api } from '../api/client'
import { humanizeError } from '../lib/errors'
import { CATALOG_IMPACT_TEXT as T, WORKFLOW_STATUS_LABEL } from '../lib/terms'
import type { CatalogImpact, CatalogImpactNode, CatalogImpactTemplate } from '../types'
import { Skeleton } from './ui'
// ===========================================================================
// 影响面:哪些已发布、受管的模板引用这张表(GET /catalog/{table}/impact)。数据目录的表详情里一栏,助手的
// 「建议更新数据目录」卡片保存之后也列一遍——改了目录,人要知道哪些正式出具的数可能跟着变。
//
// 「直接引用」和「可能涉及」分开写:前者 SQL 里写着这张表,后者是 Agent 运行时自己写 SQL,静态看不出。
// 合并查询写明经由哪个输入。模板名点进去是画布(定位到那个节点)。
// ===========================================================================
const SHOWN = 5
export function CatalogImpactList({ sourceId, table, refreshKey, compact = false }: {
sourceId: string
table: string
/** 变了就重新统计(表详情传目录版本:保存之后刷新) */
refreshKey?: number | string
/** 卡片里的紧凑版:只列前几个模板,节点收成一行 */
compact?: boolean
}) {
const [data, setData] = useState<CatalogImpact | null>(null)
const [error, setError] = useState<unknown>(null)
const [all, setAll] = useState(false)
const seq = useRef(0)
const load = useCallback(() => {
const mine = ++seq.current
setError(null)
api.dataCatalog.impact(sourceId, table)
.then((d) => { if (mine === seq.current) setData(d) })
.catch((e) => { if (mine === seq.current) setError(e) })
}, [sourceId, table])
useEffect(() => {
setData(null)
setAll(false)
load()
}, [load, refreshKey])
if (error && !data) {
return (
<p className="flex flex-wrap items-center gap-2 text-2xs text-[var(--err)]" data-catalog-impact="error">
{T.error}:{humanizeError(error).title}
<button type="button" className="rounded px-1 text-dim hover:bg-hover hover:text-fg" onClick={load}>{T.retry}</button>
</p>
)
}
if (!data) return <div data-catalog-impact="loading"><Skeleton rows={compact ? 1 : 2} height={compact ? 12 : 20} gap={6} /></div>
const list = data.templates
if (!list.length) {
return (
<p className={clsx('text-faint', compact ? 'text-2xs' : 'rounded-lg border px-3 py-4 text-xs')} data-catalog-impact="0">
{compact ? T.afterSaveNone : T.none}
</p>
)
}
const shown = all || !compact ? list : list.slice(0, SHOWN)
return (
<div data-catalog-impact={list.length}>
{compact && <p className="mb-1 text-2xs text-dim">{T.afterSave(list.length)}</p>}
<ul className={clsx(compact ? 'space-y-0.5' : 'divide-y divide-[var(--hairline)] rounded-lg border bg-panel')}>
{shown.map((t) => <TemplateRow key={t.workflow_id} t={t} compact={compact} />)}
</ul>
{compact && list.length > shown.length && (
<button type="button" className="mt-0.5 rounded px-1 text-2xs text-dim hover:bg-hover hover:text-fg" onClick={() => setAll(true)}>
{T.more(list.length - shown.length)}
</button>
)}
</div>
)
}
function ImpactBadge({ impact }: { impact: 'direct' | 'possible' }) {
return (
<span className={clsx('chip shrink-0', impact === 'direct' ? 'text-fg' : 'text-faint')} title={T.impactHint[impact]}
style={impact === 'direct' ? { borderColor: 'var(--st-waiting)', color: 'var(--st-waiting)' } : undefined}
data-impact={impact}>
{T.impact[impact]}
</span>
)
}
function nodeText(n: CatalogImpactNode): string {
const who = n.member ? `「${n.label}」${T.member(n.member)}` : `「${n.label}」`
const via = n.via?.length ? `(${T.via(n.via.map((v) => `「${v.label}」`).join('、'))})` : ''
return `${who}${via}`
}
function TemplateRow({ t, compact }: { t: CatalogImpactTemplate; compact: boolean }) {
const first = t.nodes.find((n) => n.impact === t.impact) ?? t.nodes[0]
const href = `/studio/${encodeURIComponent(t.workflow_id)}${first ? `?focus=${encodeURIComponent(first.node_id)}` : ''}`
return (
<li className={clsx('min-w-0', compact ? 'text-2xs' : 'px-3 py-2 text-xs')} data-impact-template={t.workflow_id} data-impact-level={t.impact}>
<div className="flex min-w-0 flex-wrap items-center gap-x-2 gap-y-0.5">
<GitBranch size={compact ? 10 : 11} className="shrink-0 text-faint" aria-hidden />
<Link to={href} className="min-w-0 truncate font-medium hover:underline">{t.name}</Link>
<span className="mono shrink-0 text-2xs text-faint">{T.version(t.version)}</span>
<span className="shrink-0 text-2xs text-faint">{WORKFLOW_STATUS_LABEL[t.level] ?? t.level}</span>
<ImpactBadge impact={t.impact} />
</div>
{!compact && (
<ul className="mt-1 space-y-0.5 pl-[19px] text-2xs text-dim">
{t.nodes.map((n) => (
<li key={`${n.node_id}:${n.member ?? ''}`} className="flex min-w-0 items-baseline gap-1.5" data-impact-node={n.node_id}>
<span className="min-w-0 break-words">{nodeText(n)}</span>
<span className={clsx('shrink-0', n.impact === 'direct' ? 'text-dim' : 'text-faint')}>{T.impact[n.impact]}</span>
</li>
))}
</ul>
)}
</li>
)
}
+24
View File
@@ -2187,3 +2187,27 @@ export const CATALOG_PATCH_TEXT = {
whereRelationNew: (to: string) => `新增指向 ${to} 的关联关系`,
sourceOf: (source: string) => `数据源「${source}」`,
}
/**
* 影响面:哪些已发布、受管的模板引用这张表。按每个模板当前的已发布版本统计(正式运行用的是它)。
* 「直接引用」:查询的 SQL 里写着这张表;「可能涉及」:Agent 绑定了这个数据源的查询工具,SQL 运行时才生成。
*/
export const CATALOG_IMPACT_TEXT = {
title: '引用这张表的模板',
hint: '按每个模板当前的已发布版本统计',
none: '没有已发布或受管的模板引用这张表',
count: (n: number) => `${formatNumber(n)} 个模板`,
impact: { direct: '直接引用', possible: '可能涉及' } as Record<'direct' | 'possible', string>,
impactHint: {
direct: '查询的 SQL 中写着这张表',
possible: 'Agent 绑定了这个数据源的查询工具,SQL 在运行时生成,可能用到这张表',
} as Record<'direct' | 'possible', string>,
version: (v: number) => `v${v}`,
via: (labels: string) => `经由输入${labels}`,
member: (name: string) => `成员「${name}」`,
afterSave: (n: number) => `${formatNumber(n)} 个已发布模板引用这张表,下次正式运行时会提示数据目录有变化:`,
afterSaveNone: '没有已发布或受管的模板引用这张表',
more: (n: number) => `另有 ${formatNumber(n)} 个`,
error: '无法统计引用这张表的模板',
retry: '重试',
}
+11 -2
View File
@@ -11,9 +11,10 @@ import { DeleteButton, EmptyState, ErrorState, Notice, Skeleton, Spinner, confir
import { useLeaveGuard } from '../../lib/leave'
import { formatDateTime, formatNumber } from '../../lib/format'
import {
CATALOG_CARDINALITY_LABEL, CATALOG_KIND_HINT, CATALOG_KIND_LABEL, CATALOG_TABLE_FIELD_HINT, CATALOG_TABLE_FIELD_LABEL,
CATALOG_TEXT as CT,
CATALOG_CARDINALITY_LABEL, CATALOG_IMPACT_TEXT, CATALOG_KIND_HINT, CATALOG_KIND_LABEL, CATALOG_TABLE_FIELD_HINT,
CATALOG_TABLE_FIELD_LABEL, CATALOG_TEXT as CT,
} from '../../lib/terms'
import { CatalogImpactList } from '../../components/CatalogImpact'
import { CountBadges, ItemMark, StatusChip, StatusLegend, SystemNotesNotice } from './parts'
import type { ReviewTarget } from './parts'
import { ColumnTable, Value } from './ColumnTable'
@@ -382,6 +383,14 @@ export function TableDetail({ sourceId, table, rows, drafted, prev, next, onOpen
onChange={setRelations} />
: <RelationList relations={notes.relations ?? []} busy={!!busy} targetOf={targetOf} onOpen={onOpen} onReview={review} />}
</section>
<section aria-labelledby="catalog-sec-impact" data-catalog-section="impact">
<div className="mb-2 flex flex-wrap items-baseline gap-2">
<h3 id="catalog-sec-impact" className="text-xs font-semibold">{CATALOG_IMPACT_TEXT.title}</h3>
<span className="text-2xs text-faint">{CATALOG_IMPACT_TEXT.hint}</span>
</div>
<CatalogImpactList sourceId={sourceId} table={detail.table_name} refreshKey={detail.version} />
</section>
</div>
</div>
+3 -1
View File
@@ -4,6 +4,7 @@ import { BookMarked, Check, ExternalLink, RotateCw } from 'lucide-react'
import clsx from 'clsx'
import { ApiError, api } from '../api/client'
import { Notice, Spinner, toast } from '../components/ui'
import { CatalogImpactList } from '../components/CatalogImpact'
import { CATALOG_PATCH_TEXT as T, CATALOG_STATUS_LABEL } from '../lib/terms'
import type { CatalogPatchChange } from '../types'
import {
@@ -15,7 +16,7 @@ import {
// 改后和理由,人点「保存到数据目录」才写入——服务端从不自动写。
//
// 保存带着提案对照的版本:别人在这之后改过这张表就 409,卡片写明「这张表刚被修改过」,「重新载入」按最新的
// 目录重算改前和改后(预览接口,只读),人看过再保存。保存成功后给到数据目录那张表的入口。
// 目录重算改前和改后(预览接口,只读),人看过再保存。保存成功后列出受影响的模板(影响面),给到数据目录那张表的入口。
//
// 卡片的结局(已保存、已忽略、重新载入后的内容)记在模块级的表里:画布右栏在对话层和运行层之间切换会卸载
// 卡片,回来时不能又变回「未保存」,再点一次只会撞上 409。
@@ -151,6 +152,7 @@ export function CatalogPatchCard({ patch, turnId, dense = false }: {
{T.open} <ExternalLink size={10} aria-hidden />
</Link>
</p>
{saved && <CatalogImpactList sourceId={patch.sourceId} table={patch.table} compact />}
</div>
)
: (
+31
View File
@@ -2739,3 +2739,34 @@ export interface CatalogPatchPreview {
/** 不合法的项的说明(目录刚被改过、列已删除……) */
problems: string[]
}
/** 影响面里模板的一个节点(服务端 catalog_impact.template_refs) */
export interface CatalogImpactNode {
node_id: string
label: string
type: string
/** direct:SQL 里写着这张表;possible:Agent 运行时自己写 SQL,可能用到 */
impact: 'direct' | 'possible'
/** 合并查询:经由哪些输入(别名 → 节点) */
via?: { node_id: string; label: string; alias: string }[]
/** 协作节点:绑定了查询工具的成员 */
member?: string
}
/** 引用这张表的一个已发布或受管模板(按它当前的已发布版本算) */
export interface CatalogImpactTemplate {
workflow_id: string
name: string
/** 当前的已发布版本 */
version: number
level: 'published' | 'governed'
impact: 'direct' | 'possible'
nodes: CatalogImpactNode[]
}
/** GET /datasources/{id}/catalog/{table}/impact */
export interface CatalogImpact {
table: string
/** 直接引用的在前 */
templates: CatalogImpactTemplate[]
}
+48
View File
@@ -395,6 +395,11 @@ function catalogHandlers(state, replies = {}) {
json({ tables: state.tables[parts(key)[2]].filter((t) => !t.missing).map((t) => t.name), summary: '', synced_at: T0 })(route)],
[/^GET \/datasources\/check-cat-[a-z]+\/catalog$/, (route, { key }) => json(listOf(state, parts(key)[2]))(route)],
[/^POST \/datasources\/check-cat-[a-z]+\/catalog\/draft$/, reply('draft', draftReply(state))],
// 影响面:引用这张表的已发布、受管模板(只读)。没配的表答空
[/^GET \/datasources\/check-cat-[a-z]+\/catalog\/[^/]+\/impact$/, reply('impact', (route, { key }) => {
const name = parts(key)[4]
return json({ table: name, templates: state.impact?.[name] ?? [] })(route)
})],
[/^GET \/datasources\/check-cat-[a-z]+\/catalog\/[^/]+$/, reply('get', (route, { key }) => {
const { src, t } = findTable(state, key)
return t ? json(detailOf(src, t))(route) : json({ detail: `数据源中没有表 ${parts(key)[4]},可能已被删除或尚未探查结构` }, 404)(route)
@@ -775,6 +780,49 @@ await section('数据目录 · 表详情与单项审阅', async () => {
await close()
})
await section('数据目录 · 引用这张表的模板(影响面)', async () => {
const state = freshState()
state.impact = {
visits: [
{ workflow_id: 'wf-daily', name: '入园日报', version: 4, level: 'governed', impact: 'direct', nodes: [
{ node_id: 'q', label: '查询入园人数', type: 'tool', impact: 'direct' },
{ node_id: 'm', label: '合并入园和订单', type: 'merge', impact: 'direct', via: [{ node_id: 'q', label: '查询入园人数', alias: 'v' }] },
] },
{ workflow_id: 'wf-ask', name: '客流分析', version: 2, level: 'published', impact: 'possible', nodes: [
{ node_id: 'ask', label: '分析客流', type: 'agent', impact: 'possible' },
] },
],
}
const { page, sent, errors, close } = await open(`/data/catalog/${S1}`, { handlers: catalogHandlers(state) })
await openDetail(page, S1, 'visits')
const sec = page.locator('[data-catalog-section="impact"]')
await sec.locator('[data-catalog-impact="2"]').waitFor({ timeout: 3000 })
const text = await sec.innerText()
check('表详情有「引用这张表的模板」一栏,写明按已发布版本统计', text.includes('引用这张表的模板') && text.includes('按每个模板当前的已发布版本统计'))
const daily = sec.locator('[data-impact-template="wf-daily"]')
check('直接引用:模板名、版本、受管、「直接引用」', ['入园日报', 'v4', '受管', '直接引用'].every((x) => text.includes(x))
&& await daily.getAttribute('data-impact-level') === 'direct')
check('合并查询写明经由哪个输入', (await daily.locator('[data-impact-node="m"]').innerText()).includes('经由输入「查询入园人数」'))
check('Agent 写「可能涉及」,悬停说明原因', (await sec.locator('[data-impact-template="wf-ask"] [data-impact="possible"]').getAttribute('title')).includes('运行时生成'))
check('模板名点进去是画布,定位到那个节点', (await daily.locator('a').first().getAttribute('href')) === '/studio/wf-daily?focus=q')
// 开发模式下 StrictMode 会把挂载时的取数做两遍,只查有没有取、没有写
check('统计影响面只读', writes(sent, /impact/).length === 0 && count(sent, /\/visits\/impact$/) >= 1,
`${count(sent, /\/visits\/impact$/)} 次`)
// 没有模板引用的表:直说没有
await openDetail(page, S1, 'orders')
check('没有模板引用:直说没有', (await page.locator('[data-catalog-section="impact"] [data-catalog-impact="0"]').innerText())
.includes('没有已发布或受管的模板引用这张表'))
// 表详情保存之后(版本变了)重新统计
const before = count(sent, /\/orders\/impact$/)
await page.locator('[data-catalog-edit]').click()
await page.locator('[data-input="grain"]').fill('一笔订单一行(检查)')
await page.locator('[data-catalog-save]').click()
await until(async () => count(sent, /\/orders\/impact$/) > before, 3000)
check('保存之后重新统计影响面', count(sent, /\/orders\/impact$/) > before, `${before} → ${count(sent, /\/orders\/impact$/)}`)
check('没有运行时报错', errors.length === 0, errors.join(' | '))
await close()
})
await section('数据目录 · 批量确认', async () => {
const state = freshState()
const { page, sent, errors, natives, close } = await open(`/data/catalog/${S1}`, { handlers: catalogHandlers(state) })
+10 -1
View File
@@ -795,7 +795,12 @@ await section('建议更新数据目录:卡片的保存、忽略和 409(cata
return route.fulfill({ json: detail(5) })
})
await c.page.route('**/api/datasources/src-scenic/catalog/visits/impact', (route) =>
route.fulfill({ json: { table: 'visits', templates: [] } }))
route.fulfill({ json: { table: 'visits', templates: [
{ workflow_id: 'wf-daily', name: '入园日报', version: 4, level: 'governed', impact: 'direct',
nodes: [{ node_id: 'q', label: '查询入园人数', type: 'tool', impact: 'direct' }] },
{ workflow_id: 'wf-ask', name: '客流分析', version: 2, level: 'published', impact: 'possible',
nodes: [{ node_id: 'ask', label: '分析客流', type: 'agent', impact: 'possible' }] },
] } }))
const card = c.page.locator('[data-catalog-patch="visits"]')
check('卡片在:标题写「建议更新数据目录」', await card.count() === 1 && (await card.innerText()).includes('建议更新数据目录'))
const codes = card.locator('[data-patch-change="columns.status.codes"]')
@@ -830,6 +835,10 @@ await section('建议更新数据目录:卡片的保存、忽略和 409(cata
check('保存成功:写明第几版,给到数据目录的入口', (await card.locator('[data-patch-done]').innerText()).includes('已保存到数据目录(第 5 版)')
&& (await card.locator('[data-patch-open]').getAttribute('href')) === '/data/catalog/src-scenic/visits')
check('保存后不再有保存按钮', await card.locator('[data-patch-save]').count() === 0)
await card.locator('[data-catalog-impact="2"]').waitFor({ timeout: 3000 })
const impact = await card.locator('[data-catalog-impact]').innerText()
check('保存后列出受影响的模板:直接引用、可能涉及', impact.includes('2 个已发布模板引用这张表') && impact.includes('入园日报')
&& impact.includes('直接引用') && impact.includes('可能涉及'), impact.replace(/\s+/g, ' '))
check('没有运行时报错', c.errors.length === 0, c.errors.join(' | '))
await c.page.close()