mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 02:11:58 +00:00
合入数据目录阶段 1A：目录存储、接口、外键与命名推断、起草、冻结进快照、db_schema 带目录
This commit is contained in:
ThearchyHelios
@@ -0,0 +1,226 @@
|
||||
"""数据目录接口：/api/datasources/{source_id}/catalog。
|
||||
|
||||
- GET /catalog 表清单（没有目录的表也列出来）
|
||||
- GET /catalog/{table} 单表目录 + 表结构
|
||||
- PUT /catalog/{table} 整份提交（乐观锁），改动过的项记为人工确认
|
||||
- POST /catalog/{table}/review 单项审阅：确认、驳回、撤销审阅
|
||||
- POST /catalog/draft 同步起草（注释、外键、命名推断，可选模型）
|
||||
|
||||
规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径
|
||||
不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。
|
||||
|
||||
上传表格的源按当前快照的表结构回答（table_versions.resolve_source），和查询实际用的库是同一个版本。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Literal
|
||||
|
||||
from fastapi import APIRouter, Depends, Header, HTTPException
|
||||
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.db.base import get_session
|
||||
from app.db.models import DataSource
|
||||
|
||||
router = APIRouter(prefix="/api/datasources", tags=["catalog"])
|
||||
|
||||
|
||||
class CatalogPutIn(BaseModel):
|
||||
notes: dict[str, Any]
|
||||
#: 读到的版本；这张表还没有目录时为 0
|
||||
if_version: int = Field(ge=0)
|
||||
|
||||
|
||||
class CatalogReviewIn(BaseModel):
|
||||
#: grain、columns.amount.measure、relations.<编号>
|
||||
path: str = Field(min_length=1, max_length=600)
|
||||
action: Literal["confirm", "reject", "reset"]
|
||||
if_version: int = Field(ge=0)
|
||||
|
||||
|
||||
class CatalogDraftIn(BaseModel):
|
||||
#: 要起草的表；不给或为空时按使用次数取前 20 张
|
||||
tables: list[str] | None = Field(default=None, max_length=200)
|
||||
#: 是否请助手的模型起草中文名、粒度、列的含义等
|
||||
use_model: bool = False
|
||||
|
||||
|
||||
async def _resolved(session: AsyncSession, source_id: str) -> tuple[DataSource, Any]:
|
||||
"""(数据源记录, 用来读表结构的源)：上传源解析成绑定当前快照的视图。"""
|
||||
row = await session.get(DataSource, source_id)
|
||||
if row is None:
|
||||
raise HTTPException(404, "数据源不存在，可能已被删除")
|
||||
try:
|
||||
return row, await table_versions.resolve_source(session, row)
|
||||
except table_versions.SnapshotMissing as e:
|
||||
raise HTTPException(409, str(e)) from e
|
||||
|
||||
|
||||
def _iso(value: Any) -> str | None:
|
||||
return value.isoformat() if value is not None else None
|
||||
|
||||
|
||||
def _item_value(notes: dict[str, Any], name: str) -> tuple[Any, str | None]:
|
||||
item = notes.get(name)
|
||||
if not isinstance(item, dict) or item.get("status") == "rejected":
|
||||
return None, None
|
||||
return item.get("value"), item.get("status")
|
||||
|
||||
|
||||
def _summary_row(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogEntry | None,
|
||||
usage: int) -> dict[str, Any]:
|
||||
notes = entry.notes if entry else {}
|
||||
label, label_status = _item_value(notes, "label")
|
||||
kind, _ = _item_value(notes, "kind")
|
||||
relations = notes.get("relations") if isinstance(notes.get("relations"), list) else []
|
||||
return {
|
||||
"table_name": name,
|
||||
"qualified": (meta or {}).get("qualified", name),
|
||||
"is_view": bool((meta or {}).get("is_view")),
|
||||
# 表结构里已经没有这张表（被删了、改名了），目录还留着：界面据此提示清理
|
||||
"in_schema": meta is not None,
|
||||
"label": label,
|
||||
"label_status": label_status,
|
||||
"kind": kind,
|
||||
"counts": catalog.status_counts(notes),
|
||||
"relations": sum(1 for r in relations if isinstance(r, dict) and r.get("status") != "rejected"),
|
||||
"usage": usage,
|
||||
"version": entry.version if entry else 0,
|
||||
"updated_at": _iso(entry.updated_at) if entry else None,
|
||||
"updated_by": entry.updated_by if entry else None,
|
||||
}
|
||||
|
||||
|
||||
def _structure(meta: dict[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
"qualified": meta.get("qualified"),
|
||||
"is_view": bool(meta.get("is_view")),
|
||||
"comment": meta.get("comment"),
|
||||
"columns": introspect.table_columns(meta),
|
||||
"primary_key": list(meta.get("primary_key") or []),
|
||||
# 升级前探查的缓存里没有这两项：为 null 表示「不知道」，和空列表（读过、没有）分开
|
||||
"foreign_keys": meta.get("foreign_keys"),
|
||||
"unique": meta.get("unique"),
|
||||
}
|
||||
|
||||
|
||||
async def _detail(session: AsyncSession, source: Any, table: str, entry: catalog.CatalogEntry | None,
|
||||
usage: int | None = None) -> dict[str, Any]:
|
||||
meta = ((source.schema_cache or {}).get("tables") or {}).get(table)
|
||||
if usage is None:
|
||||
usage = (await catalog.table_usage(session, source)).get(table, 0)
|
||||
return {
|
||||
"table_name": table,
|
||||
"in_schema": meta is not None,
|
||||
"notes": entry.notes if entry else {},
|
||||
"version": entry.version if entry else 0,
|
||||
"updated_at": _iso(entry.updated_at) if entry else None,
|
||||
"updated_by": entry.updated_by if entry else None,
|
||||
"structure": _structure(meta) if meta is not None else None,
|
||||
"system_notes": catalog.system_notes_source(source),
|
||||
"usage": usage,
|
||||
}
|
||||
|
||||
|
||||
async def _table_key(session: AsyncSession, source: Any, table: str) -> tuple[str, catalog.CatalogEntry | None]:
|
||||
"""路径里的表名 → 目录的键。表结构里有的按表结构的写法；表结构里没有、但还留着目录的也认（便于清理）。"""
|
||||
key = catalog.resolve_table_name(source.schema_cache, table)
|
||||
entry = await catalog.read_entry(session, source.id, key or table)
|
||||
if key is None and entry is None:
|
||||
raise HTTPException(404, f"数据源中没有表 {table}，可能已被删除或尚未探查结构")
|
||||
return key or table, entry
|
||||
|
||||
|
||||
def _conflict(e: catalog.CatalogConflict) -> HTTPException:
|
||||
return HTTPException(409, str(e))
|
||||
|
||||
|
||||
def _invalid(e: catalog.CatalogInvalid) -> HTTPException:
|
||||
shown = "；".join(e.problems[:8]) + (f"；另有 {len(e.problems) - 8} 处" if len(e.problems) > 8 else "")
|
||||
return HTTPException(422, f"数据目录的格式不正确：{shown}")
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog")
|
||||
async def list_catalog(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""表清单：表结构里的每张表各一行（没有目录的也列），使用次数多的在前；目录还在、表已不在的排最后。"""
|
||||
_, source = await _resolved(session, source_id)
|
||||
cache = source.schema_cache or {}
|
||||
tables = cache.get("tables") or {}
|
||||
entries = await catalog.read_catalog(session, source.id)
|
||||
usage = await catalog.table_usage(session, source)
|
||||
order = {name: i for i, name in enumerate(tables)}
|
||||
rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0)) for name, meta in tables.items()]
|
||||
rows.sort(key=lambda r: (-r["usage"], order[r["table_name"]]))
|
||||
rows += [_summary_row(name, None, entry, 0) for name, entry in sorted(entries.items()) if name not in tables]
|
||||
return {
|
||||
"tables": rows,
|
||||
"system_notes": catalog.system_notes_source(source),
|
||||
# 没有表结构时说明原因（尚未探查 / 探查失败），界面据此引导去探查
|
||||
"schema_note": None if tables else introspect.why_empty(cache),
|
||||
}
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/{table}")
|
||||
async def get_catalog_table(source_id: str, table: str,
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
_, source = await _resolved(session, source_id)
|
||||
key, entry = await _table_key(session, source, table)
|
||||
return await _detail(session, source, key, entry)
|
||||
|
||||
|
||||
@router.put("/{source_id}/catalog/{table}")
|
||||
async def put_catalog_table(source_id: str, table: str, payload: CatalogPutIn,
|
||||
x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""整份提交一张表的目录。改动过的项记为人工确认；版本不符 409。"""
|
||||
_, source = await _resolved(session, source_id)
|
||||
key, _ = await _table_key(session, source, table)
|
||||
try:
|
||||
entry = await catalog.save_human_edit(session, source.id, key, payload.notes, if_version=payload.if_version,
|
||||
actor=actor_of(x_actor))
|
||||
except catalog.CatalogConflict as e:
|
||||
raise _conflict(e) from e
|
||||
except catalog.CatalogInvalid as e:
|
||||
raise _invalid(e) from e
|
||||
return await _detail(session, source, key, entry)
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/draft")
|
||||
async def draft_catalog(source_id: str, payload: CatalogDraftIn,
|
||||
x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""同步起草。返回每张表新增、更新、删除了几项；模型用不了或某张表的模型起草失败，照实写在结果里。"""
|
||||
_, source = await _resolved(session, source_id)
|
||||
cache = source.schema_cache or {}
|
||||
if not cache.get("tables"):
|
||||
raise HTTPException(409, f"{introspect.why_empty(cache)}，无法起草数据目录。请先在数据源卡片上点「探查结构」")
|
||||
report = await catalog.draft_catalog(session, source, tables=payload.tables or None,
|
||||
use_model=payload.use_model, actor=actor_of(x_actor))
|
||||
rows = [{"table_name": r.table, "added": r.added, "updated": r.updated, "removed": r.removed,
|
||||
"version": r.version, "error": r.error, "model_error": r.model_error} for r in report.tables]
|
||||
return {
|
||||
"tables": rows,
|
||||
"total": {k: sum(r[k] for r in rows) for k in ("added", "updated", "removed")},
|
||||
"model_used": report.model_used,
|
||||
"model": report.model or None,
|
||||
"model_error": report.model_error,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/{table}/review")
|
||||
async def review_catalog_item(source_id: str, table: str, payload: CatalogReviewIn,
|
||||
x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""单项审阅。版本不符 409；路径不对、项不存在 422。"""
|
||||
_, source = await _resolved(session, source_id)
|
||||
key, _ = await _table_key(session, source, table)
|
||||
try:
|
||||
entry = await catalog.review_entry(session, source.id, key, payload.path, payload.action,
|
||||
if_version=payload.if_version, actor=actor_of(x_actor))
|
||||
except catalog.CatalogConflict as e:
|
||||
raise _conflict(e) from e
|
||||
except catalog.CatalogPathError as e:
|
||||
raise HTTPException(422, str(e)) from e
|
||||
return await _detail(session, source, key, entry)
|
||||
@@ -22,6 +22,7 @@ from app.api.runs import actor_of
|
||||
from app.core.errors import AUTH, NETWORK, TIMEOUT, classify, explain, raw
|
||||
from app.core.config import settings
|
||||
from app.core.crypto import encrypt, mask
|
||||
from app.data import catalog as data_catalog
|
||||
from app.data import introspect as introspect_mod
|
||||
from app.data import table_versions
|
||||
from app.data.engine import (
|
||||
@@ -805,10 +806,13 @@ async def get_schema(source_id: str, table: str | None = None,
|
||||
raise HTTPException(409, str(e)) from e
|
||||
if table:
|
||||
meta = introspect_mod.find_table(row, table)
|
||||
# 和 db_schema 工具同一个说法：有数据目录的表，字段清单后面附上目录
|
||||
entries = await data_catalog.read_catalog(session, found.id)
|
||||
return {
|
||||
"table": table,
|
||||
# 给模型看的那段文本照旧；界面画表格用下面的结构化字段
|
||||
"detail": introspect_mod.describe_table(row, table),
|
||||
"detail": introspect_mod.describe_table(row, table,
|
||||
catalog={n: e.notes for n, e in entries.items()}),
|
||||
"found": meta is not None,
|
||||
"qualified": meta.get("qualified", table) if meta else None,
|
||||
"kind": ("view" if meta.get("is_view") else "table") if meta else None,
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -14,6 +14,7 @@
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
@@ -118,8 +119,58 @@ def _target_schema(source: Any, explicit: str | None) -> str | None:
|
||||
return str(configured) if configured else None
|
||||
|
||||
|
||||
def _foreign_keys(raw: list[dict[str, Any]], target_schema: str | None,
|
||||
default_schema: str | None) -> list[dict[str, Any]]:
|
||||
"""反射出来的外键 → [{columns, to_table, to_columns}]。
|
||||
|
||||
to_table 和 schema_cache["tables"] 的键同一个口径：指向同一个 schema 的只写表名；跨 schema 的写
|
||||
「schema.表名」（那张表不在这次探查的结果里，关系照记，图里是一个悬空的端点）。to_columns 可能为空
|
||||
（SQLite 的 REFERENCES t 省略了列），由数据目录按被指向表的主键补。
|
||||
"""
|
||||
home = {s for s in (target_schema, default_schema) if s}
|
||||
out: list[dict[str, Any]] = []
|
||||
for fk in raw:
|
||||
cols = [c for c in fk.get("constrained_columns") or [] if c]
|
||||
to_table = fk.get("referred_table")
|
||||
if not cols or not to_table:
|
||||
continue
|
||||
ref_schema = fk.get("referred_schema")
|
||||
out.append({
|
||||
"columns": cols,
|
||||
"to_table": to_table if not ref_schema or ref_schema in home else f"{ref_schema}.{to_table}",
|
||||
"to_columns": [c for c in fk.get("referred_columns") or [] if c],
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def _unique_groups(constraints: list[dict[str, Any]], indexes: list[dict[str, Any]],
|
||||
pk: list[str]) -> list[list[str]]:
|
||||
"""唯一约束和唯一索引的列组，去重，不含和主键相同的那组。
|
||||
|
||||
同一组唯一列在各家可能出现两次：MySQL 的 UNIQUE 约束本身就是唯一索引，PostgreSQL 的唯一约束背后也有
|
||||
一个索引，Oracle 的主键背后还有一个唯一索引。表达式索引（列名里有 None）不算。
|
||||
"""
|
||||
seen = {tuple(c.lower() for c in pk)} if pk else set()
|
||||
groups: list[list[str]] = []
|
||||
candidates = [u.get("column_names") for u in constraints]
|
||||
candidates += [ix.get("column_names") for ix in indexes if ix.get("unique")]
|
||||
for cols in candidates:
|
||||
if not cols or any(c is None for c in cols):
|
||||
continue
|
||||
key = tuple(str(c).lower() for c in cols)
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
groups.append([str(c) for c in cols])
|
||||
return groups
|
||||
|
||||
|
||||
async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any]:
|
||||
"""探查整个库的结构，结果直接存进 DataSource.schema_cache。"""
|
||||
"""探查整个库的结构，结果直接存进 DataSource.schema_cache。
|
||||
|
||||
每张表：列（名字、类型、可空、注释）、主键、表注释、是不是视图；读得到的话还有外键约束
|
||||
（foreign_keys）和唯一约束 / 唯一索引（unique）——读不到的表不写这两个键，探查照常完成。
|
||||
"""
|
||||
engine = await engines.get(source)
|
||||
target = _target_schema(source, schema)
|
||||
|
||||
@@ -151,9 +202,14 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any
|
||||
failures: dict[str, str] = {}
|
||||
|
||||
def _multi(fn: Any, single: Any, names: list[str], *,
|
||||
errors: dict[str, str] | None = None) -> dict[Any, Any]:
|
||||
errors: dict[str, str] | None = None, ok: set[str] | None = None) -> dict[Any, Any]:
|
||||
"""批量反射，失败时逐个对象重试。ok 给了的话，记下真正读到了的对象名：批量成功时是全部
|
||||
（没有约束的表批量接口也会给空列表，或者干脆不给），逐个重试时只有没报错的那些。"""
|
||||
try:
|
||||
return dict(fn(schema=target_schema, filter_names=names, kind=ObjectKind.ANY))
|
||||
got = dict(fn(schema=target_schema, filter_names=names, kind=ObjectKind.ANY))
|
||||
if ok is not None:
|
||||
ok.update(names)
|
||||
return got
|
||||
except NotImplementedError:
|
||||
return {} # 表注释这类不是所有方言都支持
|
||||
except Exception: # noqa: BLE001 - 批量取不到，逐个对象再试
|
||||
@@ -162,6 +218,8 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any
|
||||
for name in names:
|
||||
try:
|
||||
out[(target_schema, name)] = single(name, schema=target_schema)
|
||||
if ok is not None:
|
||||
ok.add(name)
|
||||
except NotImplementedError:
|
||||
break
|
||||
except Exception as e: # noqa: BLE001 - 只跳过这一个对象
|
||||
@@ -176,6 +234,20 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any
|
||||
readable = [n for n in picked if columns_by.get((target_schema, n))]
|
||||
pk_by = _multi(inspector.get_multi_pk_constraint, inspector.get_pk_constraint, readable)
|
||||
comment_by = _multi(inspector.get_multi_table_comment, inspector.get_table_comment, readable)
|
||||
# 外键和唯一约束给数据目录起草关系用（data/catalog.py）。放在列、主键、注释之后读：PostgreSQL 上一条
|
||||
# 语句报错会让整个事务作废，先读的那几样不能被它连累。读不到的表不写这两个键（ok 集合里没有它），
|
||||
# 下游据此分得清「读过、没有」和「不知道」——升级前探查的缓存里也没有这两个键。
|
||||
fk_ok: set[str] = set()
|
||||
uq_ok: set[str] = set()
|
||||
ix_ok: set[str] = set()
|
||||
fk_by = _multi(inspector.get_multi_foreign_keys, inspector.get_foreign_keys, readable, errors={}, ok=fk_ok)
|
||||
uq_by = _multi(inspector.get_multi_unique_constraints, inspector.get_unique_constraints, readable,
|
||||
errors={}, ok=uq_ok)
|
||||
ix_by = _multi(inspector.get_multi_indexes, inspector.get_indexes, readable, errors={}, ok=ix_ok)
|
||||
try:
|
||||
default_schema = inspector.default_schema_name
|
||||
except Exception: # noqa: BLE001 - 只用来判断外键是不是指向别的 schema
|
||||
default_schema = None
|
||||
|
||||
tables: dict[str, Any] = {}
|
||||
skipped: dict[str, str] = {}
|
||||
@@ -207,6 +279,11 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any
|
||||
"comment": comment,
|
||||
"is_view": name in views,
|
||||
}
|
||||
if name in fk_ok:
|
||||
tables[name]["foreign_keys"] = _foreign_keys(fk_by.get(key) or [], target_schema, default_schema)
|
||||
if name in uq_ok:
|
||||
tables[name]["unique"] = _unique_groups(uq_by.get(key) or [],
|
||||
(ix_by.get(key) or []) if name in ix_ok else [], pk)
|
||||
payload: dict[str, Any] = {"tables": tables, "truncated": truncated,
|
||||
"total": len(names) + len(views)}
|
||||
if failures and not tables:
|
||||
@@ -345,8 +422,14 @@ def table_columns(meta: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
]
|
||||
|
||||
|
||||
def describe_table(source: Any, table: str) -> str:
|
||||
"""单表的完整字段说明。agent 调 db_schema 工具时返回这个。"""
|
||||
def describe_table(source: Any, table: str, *, catalog: Mapping[str, Mapping[str, Any]] | None = None) -> str:
|
||||
"""单表的完整字段说明。agent 调 db_schema 工具时返回这个。
|
||||
|
||||
catalog 是这个源的数据目录 {表名: notes}（data/catalog.py），给了且这张表有可给模型看的内容时，在字段
|
||||
清单后面附上目录：中文名、粒度、业务主键、业务日期、有效记录条件、每列的含义 / 单位 / 度量类型 / 码值、
|
||||
关联关系，推断的项标出来。导入表格的源，表和列的说明是系统按核对结果生成的：说明照旧拼在表名和列后面，
|
||||
目录只追加说明没有覆盖的、或人工确认过的项（catalog.render_table_notes）。不给 catalog 时输出和以前一样。
|
||||
"""
|
||||
cache = source.schema_cache or {}
|
||||
tables = cache.get("tables") or {}
|
||||
# 模型可能传全名（ANALYTICS.V_TRIP_FACT），也可能只传表名，两种都认
|
||||
@@ -381,6 +464,14 @@ def describe_table(source: Any, table: str) -> str:
|
||||
marks.append(col["comment"])
|
||||
suffix = f" {'、'.join(marks)}" if marks else ""
|
||||
lines.append(f" {col['name']} {col['type']}{suffix}")
|
||||
if catalog:
|
||||
from app.data import catalog as catalog_mod
|
||||
|
||||
key = catalog_mod.resolve_table_name(cache, table)
|
||||
block = catalog_mod.render_table_notes(catalog.get(key) if key else None, meta=meta,
|
||||
system_notes=catalog_mod.system_notes_source(source))
|
||||
if block:
|
||||
lines.append(block)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
|
||||
@@ -317,6 +317,37 @@ class SnapshotActivation(Base):
|
||||
created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=utcnow)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 数据目录（data/catalog.py）
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
class CatalogNote(Base, TimestampMixin):
|
||||
"""一张表的业务目录：中文名、粒度、业务主键、业务日期、列的含义 / 单位 / 度量类型 / 码值、表间关系。
|
||||
|
||||
只记数据事实，不放计算公式（公式在口径卡里）。notes 的结构和每一项的来源、状态见 data/catalog.py。
|
||||
|
||||
一表一行而不是一项一行：起草、审阅、冻结进表结构快照都以表为单位，版本号也按表走——两个人改
|
||||
同一个源的不同表互不冲突，改同一张表时后提交的被乐观锁拦下。
|
||||
|
||||
挂外键、级联删除：目录是数据源的附属说明，源删掉了留着没有意义（和导入记录不同，后者要留作审计）。
|
||||
运行用过的那一版目录已经随表结构快照冻结进证据，不靠这张表回溯。
|
||||
"""
|
||||
|
||||
__tablename__ = "catalog_notes"
|
||||
__table_args__ = (UniqueConstraint("source_id", "table_name"),)
|
||||
|
||||
id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
|
||||
source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE"), index=True)
|
||||
# 与 schema_cache["tables"] 的键一致（不带 schema 前缀）
|
||||
table_name: Mapped[str] = mapped_column(String(255))
|
||||
notes: Mapped[dict[str, Any]] = mapped_column(default=dict)
|
||||
# 从 1 起，每次内容有变化的写入加 1；写入方必须带上读到的版本（data/catalog.write_entry）
|
||||
version: Mapped[int] = mapped_column(Integer, default=1)
|
||||
# 最近一次写入的署名（请求头 X-Actor，自报、未认证）；助手起草时可为空
|
||||
updated_by: Mapped[str | None] = mapped_column(String(100), default=None)
|
||||
|
||||
|
||||
class Setting(Base, TimestampMixin):
|
||||
"""键值形式的用户设置，前端 Settings 页直接读写。"""
|
||||
|
||||
|
||||
+2
-1
@@ -9,7 +9,7 @@ from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from app.api import (
|
||||
artifacts, conversations, copilot, datasources, evidence, governance, knowledge, runs,
|
||||
artifacts, catalog, conversations, copilot, datasources, evidence, governance, knowledge, runs,
|
||||
sandbox, settings as settings_api, source_versions, table_imports, tools, workflows,
|
||||
)
|
||||
from app.api.coded import CodedHTTPException, coded_handler
|
||||
@@ -120,6 +120,7 @@ for router in (
|
||||
# 按配方导入的 /api/datasources/imports/… 要在 datasources 的通配 /{source_id} 之前登记
|
||||
table_imports.router,
|
||||
source_versions.router,
|
||||
catalog.router,
|
||||
datasources.router,
|
||||
):
|
||||
app.include_router(router)
|
||||
|
||||
@@ -21,6 +21,7 @@ import inspect
|
||||
import json
|
||||
import logging
|
||||
import weakref
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -30,6 +31,7 @@ from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.events import EventType
|
||||
from app.data import catalog as data_catalog
|
||||
from app.data import introspect, table_versions
|
||||
from app.data.engine import SnapshotTampered, masked_columns, query_timeout, run_query
|
||||
from app.data.guard import QueryLimits, SqlRejected, is_write
|
||||
@@ -154,6 +156,8 @@ RECIPE_QUERY_HINT = " 中文表名和列名请加双引号。查单个值时，
|
||||
#: 封存事件 → schema_snapshot → 快照清单 → 各期导入清单 → 原件或清除记录，每一跳都按内容哈希
|
||||
_FROZEN_SCHEMA_KEYS = ("schema", "synced_at", "truncated", "total", "import_mode", "import_manifests",
|
||||
"snapshot_manifest")
|
||||
#: 表结构快照里放数据目录的顶层键（_store_schema）。不来自 schema_cache：目录存在 catalog_notes 表里
|
||||
CATALOG_SNAPSHOT_KEY = "catalog"
|
||||
|
||||
|
||||
def _query_description(source: Any) -> str:
|
||||
@@ -218,14 +222,31 @@ async def build_datasource_tools(
|
||||
except table_versions.SnapshotMissing as e:
|
||||
tools.extend(_unavailable_tool(row, prefix, str(e)) for prefix in prefixes)
|
||||
continue
|
||||
entries = await _catalog_of(row)
|
||||
for prefix in prefixes:
|
||||
if prefix == QUERY_PREFIX:
|
||||
tools.append(_make_query_tool(source, ctx, fixed=fixed))
|
||||
tools.append(_make_query_tool(source, ctx, fixed=fixed, catalog=entries))
|
||||
else:
|
||||
tools.append(_make_schema_tool(source, fixed=fixed))
|
||||
tools.append(_make_schema_tool(source, fixed=fixed, catalog=entries))
|
||||
return tools
|
||||
|
||||
|
||||
async def _catalog_of(row: DataSource) -> dict[str, data_catalog.CatalogEntry]:
|
||||
"""这个源的数据目录，建工具时读一次：db_schema 给模型看的和查询时冻结进表结构快照的是同一版。
|
||||
|
||||
读不到就当没有目录，工具照常可用——目录是附加说明，不能挡住查询。用单独的会话读：出错要回滚，
|
||||
而回滚会让调用方会话里已经加载的对象（row）全部过期，之后再读属性就是异步环境里的懒加载。
|
||||
"""
|
||||
from app.db.base import SessionLocal
|
||||
|
||||
try:
|
||||
async with SessionLocal() as own:
|
||||
return await data_catalog.read_catalog(own, row.id)
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("读取数据源 %s 的数据目录失败，本次不附目录", row.name)
|
||||
return {}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 上传表格的版本
|
||||
# --------------------------------------------------------------------------
|
||||
@@ -414,10 +435,12 @@ def _unavailable_tool(row: DataSource, prefix: str, reason: str) -> StructuredTo
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> StructuredTool:
|
||||
def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False,
|
||||
catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> StructuredTool:
|
||||
"""source 是手工源的 DataSource，或上传源绑定快照的 SourceView（字段同名）。
|
||||
|
||||
fixed：上传源的版本是这次运行固定的（找不到时的说法不同）。
|
||||
catalog：建工具时读到的数据目录，随表结构快照冻结（_store_schema）。
|
||||
"""
|
||||
# 数据源在 options.query_timeout_s 里配了就用它，否则取缺省。数据库按它停下语句，
|
||||
# 引擎按它告诉界面上限是多少（metadata["timeout_s"]）
|
||||
@@ -449,7 +472,7 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S
|
||||
payload["source"] = source.name
|
||||
# 表结构快照：这次查询时数据源的结构冻结下来，报告里写的表名、字段名按它核对。事后数据源
|
||||
# 重新探查、改了结构，已经跑完的运行核对的还是当时那一份。上传源冻结的是绑定快照里那份
|
||||
if schema := await _store_schema(source, ctx):
|
||||
if schema := await _store_schema(source, ctx, catalog=catalog):
|
||||
payload["schema_artifact"] = schema
|
||||
# 查询快照进工件库：出具体系的数字回指要能下钻到"这个数是哪条 SQL 查出来的"
|
||||
try:
|
||||
@@ -465,9 +488,12 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S
|
||||
extra["mask_columns"] = mask
|
||||
if version := getattr(source, "snapshot_id", None):
|
||||
extra["data_version"] = version
|
||||
# 工件引用的 meta 记下源和 SQL 里用到的表：数据目录按它数每张表被运行查询过几次
|
||||
# （catalog.table_usage），不用再逐个读工件内容。meta 不进内容哈希，快照本身一字不变
|
||||
payload["artifact"] = await put_json(
|
||||
{**payload, **extra} if extra else payload, kind="query_snapshot",
|
||||
{**payload, **extra} if extra else payload, kind=data_catalog.QUERY_SNAPSHOT_KIND,
|
||||
run_id=ctx.run_id or "", node_id=ctx.node_id or "",
|
||||
meta=data_catalog.query_snapshot_meta(source, str(payload.get("sql") or sql)),
|
||||
)
|
||||
except Exception: # noqa: BLE001 - 存不下不影响查询本身
|
||||
pass
|
||||
@@ -485,7 +511,8 @@ def _make_query_tool(source: Any, ctx: ToolContext, *, fixed: bool = False) -> S
|
||||
)
|
||||
|
||||
|
||||
async def _store_schema(source: Any, ctx: ToolContext) -> str | None:
|
||||
async def _store_schema(source: Any, ctx: ToolContext, *,
|
||||
catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> str | None:
|
||||
"""把数据源此刻的 schema_cache 存成 schema_snapshot 工件，返回工件 id。
|
||||
|
||||
上传源传进来的是绑定快照的视图，schema_cache 就是快照里冻结的那份：查的是哪一版，冻结的
|
||||
@@ -493,6 +520,10 @@ async def _store_schema(source: Any, ctx: ToolContext) -> str | None:
|
||||
|
||||
内容寻址：结构没变的话，同一次运行里查多少次都是同一件，文件只有一份。没探查过结构
|
||||
（或者探查失败、一张表都没有）返回 None：没有东西可以冻结，报告也就不核对表名。
|
||||
|
||||
数据目录一起冻结，放在顶层键 catalog 下：{表名: {version, notes（去掉驳回项）}}，只收这份结构里有的、
|
||||
有目录的表（catalog.frozen_catalog）。这次运行的模型看到的是哪一版目录，事后按快照就能回溯。已有字段
|
||||
一个不动；没有目录的源不加这个键，快照和以前一字不差；目录不变（版本和内容都不变）则快照不变。
|
||||
"""
|
||||
cache = source.schema_cache or {}
|
||||
tables = cache.get("tables")
|
||||
@@ -503,15 +534,25 @@ async def _store_schema(source: Any, ctx: ToolContext) -> str | None:
|
||||
|
||||
content = {"source": source.name, "tables": tables,
|
||||
**{k: cache[k] for k in _FROZEN_SCHEMA_KEYS if k in cache}}
|
||||
meta: dict[str, Any] = {"source": source.name, "tables": len(tables)}
|
||||
if frozen := data_catalog.frozen_catalog(catalog or {}, tables):
|
||||
content[CATALOG_SNAPSHOT_KEY] = frozen
|
||||
meta["catalog_tables"] = len(frozen)
|
||||
try:
|
||||
return await put_json(content, kind=SCHEMA_SNAPSHOT, run_id=ctx.run_id or "", node_id=ctx.node_id or "",
|
||||
meta={"source": source.name, "tables": len(tables)})
|
||||
meta=meta)
|
||||
except Exception: # noqa: BLE001 - 存不下不影响查询本身，只是这次报告不核对表名
|
||||
return None
|
||||
|
||||
|
||||
def _make_schema_tool(source: Any, *, fixed: bool = False) -> StructuredTool:
|
||||
"""source 同 _make_query_tool：上传源的表结构取绑定快照里冻结的那份。"""
|
||||
def _make_schema_tool(source: Any, *, fixed: bool = False,
|
||||
catalog: Mapping[str, data_catalog.CatalogEntry] | None = None) -> StructuredTool:
|
||||
"""source 同 _make_query_tool：上传源的表结构取绑定快照里冻结的那份。
|
||||
|
||||
catalog：建工具时读到的数据目录，查单表时附在字段清单后面（introspect.describe_table）。
|
||||
"""
|
||||
notes = {name: entry.notes for name, entry in (catalog or {}).items()}
|
||||
|
||||
async def _run(table: str | list[str] | None = None) -> str:
|
||||
if gone := _vanished(source, fixed):
|
||||
return f"{DATA_UNAVAILABLE}{gone}"
|
||||
@@ -520,7 +561,7 @@ def _make_schema_tool(source: Any, *, fixed: bool = False) -> StructuredTool:
|
||||
# 而模型两种都会写。为此拒绝一次调用，纯属浪费一步
|
||||
wanted = [t.strip() for one in wanted for t in str(one).split(",") if t.strip()]
|
||||
if wanted:
|
||||
return "\n\n".join(introspect.describe_table(source, t) for t in wanted)
|
||||
return "\n\n".join(introspect.describe_table(source, t, catalog=notes) for t in wanted)
|
||||
tables = introspect.table_names(source)
|
||||
if not tables:
|
||||
# 把实情交出去，而不是让它去点一个它点不到的按钮
|
||||
|
||||
@@ -41,3 +41,14 @@ def _prepare_schema():
|
||||
|
||||
def pytest_report_header(config) -> str:
|
||||
return f"测试数据目录：{_TMP}"
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def scenic_db(tmp_path_factory) -> str:
|
||||
"""合成的景区业务库（tests/fixtures/catalog/scenic.py）：50 张表、真外键与只能靠命名推断的关系并存。
|
||||
|
||||
每个测试进程建一次，只读使用；要改库内容的测试自己调 scenic.build(tmp_path / "x.db") 另建一个。
|
||||
"""
|
||||
from tests.fixtures.catalog import scenic
|
||||
|
||||
return str(scenic.build(tmp_path_factory.mktemp("catalog") / "scenic.db"))
|
||||
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
"""数据目录用的合成业务库夹具。
|
||||
|
||||
- scenic.py：一个虚构的「景区 + 园内门店」业务库（SQLite，50 张表 + 1 个视图），给数据目录的起草、关系图、
|
||||
表清单测试用，阶段 1B（助手按任务挑表）也用它。表名、数据全部虚构，不对应任何真实部署。
|
||||
|
||||
用法::
|
||||
|
||||
from tests.fixtures.catalog import scenic
|
||||
|
||||
path = scenic.build(tmp_path / "scenic.db") # 写一个新库，返回路径；同样的参数造出同样的内容
|
||||
scenic.FK_RELATIONS # 库里声明了外键约束的关系
|
||||
scenic.NAME_RELATIONS # 没有外键约束、只能靠列名推断出来的关系
|
||||
scenic.NOT_INFERRED # 看起来像关联、但命名推断必须放过的列（及原因）
|
||||
|
||||
conftest.py 里有个 scenic_db 夹具（每个测试进程建一次，只读使用），直接要这个参数就行。
|
||||
"""
|
||||
+30
@@ -0,0 +1,30 @@
|
||||
"""数据目录起草用的假模型：不发请求，按表名回固定的草稿。
|
||||
|
||||
model = FakeModel({"visits": {"label": "入园记录", "kind": "fact"}}, fail={"orders"})
|
||||
await catalog.draft_with_model(model, source, ["visits"])
|
||||
|
||||
- 结构化输出（with_structured_output）返回自己；ainvoke 从提示词里认出这一批有哪些表（「表 <表名>」），
|
||||
返回 {"tables": [{"table": 表名, **replies[表名]}, …]}；
|
||||
- fail 里的表所在的那一批抛异常，用来验证「一批失败只影响这一批」；
|
||||
- calls 记下每次收到的提示词全文，用来核对交给模型的内容（只有结构，没有数据）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
|
||||
class FakeModel:
|
||||
def __init__(self, replies: dict[str, dict], fail: set[str] = frozenset()) -> None:
|
||||
self.replies, self.fail, self.calls = replies, set(fail), []
|
||||
|
||||
def with_structured_output(self, schema: Any, **_: Any) -> "FakeModel":
|
||||
assert schema["title"]
|
||||
return self
|
||||
|
||||
async def ainvoke(self, messages: list[Any]) -> dict:
|
||||
text = "\n".join(str(getattr(m, "content", m)) for m in messages)
|
||||
self.calls.append(text)
|
||||
names = [t for t in self.replies if f"表 {t}\n" in text or f"表 {t}（" in text]
|
||||
if any(t in self.fail for t in names):
|
||||
raise RuntimeError("网关超时")
|
||||
return {"tables": [{"table": t, **self.replies[t]} for t in names]}
|
||||
+284
@@ -0,0 +1,284 @@
|
||||
"""虚构的景区业务库：入园记录、渠道、票种、订单、园内门店、会员、营销、停车、旅行团……共 50 张表和 1 个视图。
|
||||
|
||||
为数据目录设计的几处特点：
|
||||
|
||||
- **真外键和只能靠命名推断的关系混在一起。** 核心交易表（订单、订单明细、入园记录的票种）声明了外键；
|
||||
很多「看名字就知道连哪」的列（visits.gate_id、store_sales.productId、inventory_snapshots.storeID）没有约束，
|
||||
只能按列名推断。三种命名都有：xxx_id、xxxId、xxxID；表名有复数、驼峰（memberTags）、y→ies（categories、agencies）。
|
||||
- **关联表。** channel_visits 把渠道和入园记录连起来（复合主键，两端都是外键）：入园记录本身不带渠道，
|
||||
从 visits 到 channels 要经过它，正好是两跳的连接路径。group_visits 同理（旅行团与入园记录）。
|
||||
- **命名推断必须放过的列**（NOT_INFERRED）：自引用（employees.employee_id 是工号，不是指向自己）、
|
||||
多个候选（guide 和 guides 两张表都在，tour_assignments.guide_id 不知道指哪张）、没有对应的表
|
||||
（parking_records.lot_id、operation_logs.operator_id、categories.parentId）、类型对不上
|
||||
（feedback_forms.store_id 是文字，stores.id 是整数）。
|
||||
- **唯一约束 / 唯一索引。** orders.order_no、channels.channel_code 是列上的 UNIQUE；members.member_no、
|
||||
visits.ticket_no 是单独建的唯一索引。
|
||||
|
||||
数据是固定种子生成的假数，量很小（入园记录 1500 行），只为让查询有结果；名称一律虚构。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
DDL = """
|
||||
CREATE TABLE parks (id INTEGER PRIMARY KEY, park_code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, city TEXT,
|
||||
opened_on TEXT);
|
||||
CREATE TABLE areas (id INTEGER PRIMARY KEY, park_id INTEGER NOT NULL, name TEXT NOT NULL);
|
||||
CREATE TABLE gates (id INTEGER PRIMARY KEY, park_id INTEGER NOT NULL REFERENCES parks(id), gate_code TEXT NOT NULL,
|
||||
name TEXT);
|
||||
CREATE TABLE channels (id INTEGER PRIMARY KEY, channel_code TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
|
||||
channel_type TEXT NOT NULL);
|
||||
CREATE TABLE ticket_types (id INTEGER PRIMARY KEY, park_id INTEGER NOT NULL REFERENCES parks(id),
|
||||
type_code TEXT NOT NULL, name TEXT NOT NULL, base_price REAL NOT NULL);
|
||||
CREATE TABLE member_levels (id INTEGER PRIMARY KEY, level_name TEXT NOT NULL, discount_rate REAL);
|
||||
CREATE TABLE members (id INTEGER PRIMARY KEY, member_no TEXT NOT NULL, name TEXT, member_level_id INTEGER,
|
||||
joined_on TEXT, status INTEGER NOT NULL DEFAULT 1);
|
||||
CREATE UNIQUE INDEX ux_members_member_no ON members(member_no);
|
||||
CREATE TABLE "memberTags" (id INTEGER PRIMARY KEY, tag_name TEXT NOT NULL);
|
||||
CREATE TABLE member_tag_links (id INTEGER PRIMARY KEY, member_id INTEGER NOT NULL REFERENCES members(id),
|
||||
"memberTagId" INTEGER NOT NULL);
|
||||
CREATE TABLE orders (id INTEGER PRIMARY KEY, order_no TEXT NOT NULL UNIQUE, member_id INTEGER REFERENCES members(id),
|
||||
channel_id INTEGER NOT NULL REFERENCES channels(id), ordered_at TEXT NOT NULL,
|
||||
total_amount REAL NOT NULL, status INTEGER NOT NULL);
|
||||
CREATE TABLE order_items (id INTEGER PRIMARY KEY, order_id INTEGER NOT NULL REFERENCES orders(id),
|
||||
ticket_type_id INTEGER NOT NULL REFERENCES ticket_types(id), qty INTEGER NOT NULL,
|
||||
unit_price REAL NOT NULL, amount REAL NOT NULL);
|
||||
CREATE TABLE visits (id INTEGER PRIMARY KEY, ticket_no TEXT NOT NULL, park_id INTEGER NOT NULL REFERENCES parks(id),
|
||||
gate_id INTEGER, ticket_type_id INTEGER NOT NULL REFERENCES ticket_types(id), member_id INTEGER,
|
||||
visit_time TEXT NOT NULL, visitor_count INTEGER NOT NULL DEFAULT 1, status INTEGER NOT NULL);
|
||||
CREATE UNIQUE INDEX ux_visits_ticket_no ON visits(ticket_no);
|
||||
CREATE TABLE channel_visits (channel_id INTEGER NOT NULL REFERENCES channels(id),
|
||||
visit_id INTEGER NOT NULL REFERENCES visits(id),
|
||||
attributed_share REAL NOT NULL DEFAULT 1, PRIMARY KEY (channel_id, visit_id));
|
||||
CREATE TABLE stores (id INTEGER PRIMARY KEY, store_code TEXT NOT NULL UNIQUE, name TEXT NOT NULL,
|
||||
park_id INTEGER NOT NULL REFERENCES parks(id), area_id INTEGER, opened_on TEXT);
|
||||
CREATE TABLE categories (id INTEGER PRIMARY KEY, name TEXT NOT NULL, "parentId" INTEGER);
|
||||
CREATE TABLE products (id INTEGER PRIMARY KEY, sku TEXT NOT NULL UNIQUE, name TEXT NOT NULL, category_id INTEGER,
|
||||
list_price REAL);
|
||||
CREATE TABLE store_sales (id INTEGER PRIMARY KEY, store_id INTEGER NOT NULL REFERENCES stores(id),
|
||||
"productId" INTEGER NOT NULL, member_id INTEGER, sold_at TEXT NOT NULL, qty INTEGER NOT NULL,
|
||||
amount REAL NOT NULL);
|
||||
CREATE TABLE inventory_snapshots (id INTEGER PRIMARY KEY, "storeID" INTEGER NOT NULL, "productId" INTEGER NOT NULL,
|
||||
snapshot_date TEXT NOT NULL, on_hand INTEGER NOT NULL);
|
||||
CREATE TABLE campaigns (id INTEGER PRIMARY KEY, channel_id INTEGER REFERENCES channels(id), name TEXT NOT NULL,
|
||||
starts_on TEXT, ends_on TEXT);
|
||||
CREATE TABLE coupons (id INTEGER PRIMARY KEY, code TEXT NOT NULL UNIQUE, campaign_id INTEGER,
|
||||
discount_amount REAL NOT NULL);
|
||||
CREATE TABLE coupon_redemptions (id INTEGER PRIMARY KEY, coupon_id INTEGER NOT NULL REFERENCES coupons(id),
|
||||
order_id INTEGER NOT NULL REFERENCES orders(id), redeemed_at TEXT NOT NULL);
|
||||
CREATE TABLE payments (id INTEGER PRIMARY KEY, order_id INTEGER NOT NULL REFERENCES orders(id),
|
||||
pay_method TEXT NOT NULL, paid_amount REAL NOT NULL, paid_at TEXT NOT NULL);
|
||||
CREATE TABLE refunds (id INTEGER PRIMARY KEY, order_id INTEGER NOT NULL REFERENCES orders(id),
|
||||
refund_amount REAL NOT NULL, reason_code TEXT, refunded_at TEXT NOT NULL);
|
||||
CREATE TABLE reviews (id INTEGER PRIMARY KEY, visit_id INTEGER NOT NULL, score INTEGER NOT NULL, review_text TEXT,
|
||||
created_at TEXT NOT NULL);
|
||||
CREATE TABLE complaints (id INTEGER PRIMARY KEY, visit_id INTEGER, category_code TEXT NOT NULL,
|
||||
status INTEGER NOT NULL, created_at TEXT NOT NULL);
|
||||
CREATE TABLE events (id INTEGER PRIMARY KEY, park_id INTEGER NOT NULL REFERENCES parks(id), name TEXT NOT NULL,
|
||||
event_date TEXT NOT NULL);
|
||||
CREATE TABLE event_bookings (id INTEGER PRIMARY KEY, event_id INTEGER NOT NULL, member_id INTEGER NOT NULL,
|
||||
seats INTEGER NOT NULL, booked_at TEXT NOT NULL);
|
||||
CREATE TABLE parking_lots (id INTEGER PRIMARY KEY, park_id INTEGER NOT NULL REFERENCES parks(id),
|
||||
name TEXT NOT NULL, capacity INTEGER NOT NULL);
|
||||
CREATE TABLE parking_records (id INTEGER PRIMARY KEY, lot_id INTEGER NOT NULL, plate_masked TEXT,
|
||||
entered_at TEXT NOT NULL, exited_at TEXT, fee REAL);
|
||||
CREATE TABLE weather_daily (park_id INTEGER NOT NULL, biz_date TEXT NOT NULL, weather TEXT, temp_high REAL,
|
||||
temp_low REAL, PRIMARY KEY (park_id, biz_date));
|
||||
CREATE TABLE holidays (biz_date TEXT PRIMARY KEY, holiday_name TEXT NOT NULL, is_holiday INTEGER NOT NULL);
|
||||
CREATE TABLE dim_dates (date_key INTEGER PRIMARY KEY, biz_date TEXT NOT NULL UNIQUE, weekday INTEGER NOT NULL,
|
||||
is_weekend INTEGER NOT NULL);
|
||||
CREATE TABLE price_rules (id INTEGER PRIMARY KEY, ticket_type_id INTEGER NOT NULL REFERENCES ticket_types(id),
|
||||
valid_from TEXT NOT NULL, valid_to TEXT, price REAL NOT NULL);
|
||||
CREATE TABLE ticket_inventory (id INTEGER PRIMARY KEY, ticket_type_id INTEGER NOT NULL, biz_date TEXT NOT NULL,
|
||||
quota INTEGER NOT NULL, sold INTEGER NOT NULL);
|
||||
CREATE TABLE member_points_log (id INTEGER PRIMARY KEY, member_id INTEGER NOT NULL REFERENCES members(id),
|
||||
change_points INTEGER NOT NULL, reason TEXT, created_at TEXT NOT NULL);
|
||||
CREATE TABLE sms_logs (id INTEGER PRIMARY KEY, member_id INTEGER, template_code TEXT NOT NULL, sent_at TEXT NOT NULL);
|
||||
CREATE TABLE suppliers (id INTEGER PRIMARY KEY, name TEXT NOT NULL);
|
||||
CREATE TABLE purchase_orders (id INTEGER PRIMARY KEY, supplier_id INTEGER NOT NULL REFERENCES suppliers(id),
|
||||
store_id INTEGER NOT NULL REFERENCES stores(id), ordered_at TEXT NOT NULL,
|
||||
amount REAL NOT NULL);
|
||||
CREATE TABLE purchase_order_lines (id INTEGER PRIMARY KEY, purchase_order_id INTEGER NOT NULL,
|
||||
product_id INTEGER NOT NULL REFERENCES products(id), qty INTEGER NOT NULL,
|
||||
cost REAL NOT NULL);
|
||||
CREATE TABLE employees (id INTEGER PRIMARY KEY, employee_id TEXT NOT NULL, store_id INTEGER, name TEXT NOT NULL,
|
||||
role TEXT);
|
||||
CREATE TABLE agencies (id INTEGER PRIMARY KEY, channel_id INTEGER NOT NULL REFERENCES channels(id),
|
||||
name TEXT NOT NULL);
|
||||
CREATE TABLE tour_groups (id INTEGER PRIMARY KEY, agency_id INTEGER NOT NULL, group_size INTEGER NOT NULL,
|
||||
visit_date TEXT NOT NULL);
|
||||
CREATE TABLE group_visits (tour_group_id INTEGER NOT NULL REFERENCES tour_groups(id),
|
||||
visit_id INTEGER NOT NULL REFERENCES visits(id), PRIMARY KEY (tour_group_id, visit_id));
|
||||
CREATE TABLE guide (id INTEGER PRIMARY KEY, name TEXT);
|
||||
CREATE TABLE guides (id INTEGER PRIMARY KEY, name TEXT NOT NULL, licensed INTEGER);
|
||||
CREATE TABLE tour_assignments (id INTEGER PRIMARY KEY, tour_group_id INTEGER NOT NULL, guide_id INTEGER NOT NULL,
|
||||
assigned_on TEXT NOT NULL);
|
||||
CREATE TABLE feedback_forms (id INTEGER PRIMARY KEY, store_id TEXT, submitted_at TEXT NOT NULL, rating INTEGER);
|
||||
CREATE TABLE lost_and_found (id INTEGER PRIMARY KEY, park_id INTEGER NOT NULL, item_desc TEXT NOT NULL,
|
||||
found_at TEXT NOT NULL);
|
||||
CREATE TABLE operation_logs (id INTEGER PRIMARY KEY, operator_id INTEGER, action TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL);
|
||||
CREATE TABLE sys_config (config_key TEXT PRIMARY KEY, config_value TEXT);
|
||||
CREATE VIEW v_daily_visits AS
|
||||
SELECT park_id, substr(visit_time, 1, 10) AS biz_date, SUM(visitor_count) AS visitors
|
||||
FROM visits WHERE status = 1 GROUP BY park_id, substr(visit_time, 1, 10);
|
||||
"""
|
||||
|
||||
#: 表的个数（不含视图）
|
||||
TABLE_COUNT = 50
|
||||
VIEWS = ("v_daily_visits",)
|
||||
|
||||
#: 声明了外键约束的关系：(本表, (本表列,), 被指向的表, (被指向的列,))
|
||||
FK_RELATIONS = frozenset({
|
||||
("gates", ("park_id",), "parks", ("id",)),
|
||||
("ticket_types", ("park_id",), "parks", ("id",)),
|
||||
("member_tag_links", ("member_id",), "members", ("id",)),
|
||||
("orders", ("member_id",), "members", ("id",)),
|
||||
("orders", ("channel_id",), "channels", ("id",)),
|
||||
("order_items", ("order_id",), "orders", ("id",)),
|
||||
("order_items", ("ticket_type_id",), "ticket_types", ("id",)),
|
||||
("visits", ("park_id",), "parks", ("id",)),
|
||||
("visits", ("ticket_type_id",), "ticket_types", ("id",)),
|
||||
("channel_visits", ("channel_id",), "channels", ("id",)),
|
||||
("channel_visits", ("visit_id",), "visits", ("id",)),
|
||||
("stores", ("park_id",), "parks", ("id",)),
|
||||
("store_sales", ("store_id",), "stores", ("id",)),
|
||||
("campaigns", ("channel_id",), "channels", ("id",)),
|
||||
("coupon_redemptions", ("coupon_id",), "coupons", ("id",)),
|
||||
("coupon_redemptions", ("order_id",), "orders", ("id",)),
|
||||
("payments", ("order_id",), "orders", ("id",)),
|
||||
("refunds", ("order_id",), "orders", ("id",)),
|
||||
("events", ("park_id",), "parks", ("id",)),
|
||||
("parking_lots", ("park_id",), "parks", ("id",)),
|
||||
("price_rules", ("ticket_type_id",), "ticket_types", ("id",)),
|
||||
("member_points_log", ("member_id",), "members", ("id",)),
|
||||
("purchase_orders", ("supplier_id",), "suppliers", ("id",)),
|
||||
("purchase_orders", ("store_id",), "stores", ("id",)),
|
||||
("purchase_order_lines", ("product_id",), "products", ("id",)),
|
||||
("agencies", ("channel_id",), "channels", ("id",)),
|
||||
("group_visits", ("tour_group_id",), "tour_groups", ("id",)),
|
||||
("group_visits", ("visit_id",), "visits", ("id",)),
|
||||
})
|
||||
|
||||
#: 没有外键约束、命名推断应当找出来的关系（被指向一端都是单列主键 id）
|
||||
NAME_RELATIONS = frozenset({
|
||||
("areas", ("park_id",), "parks", ("id",)),
|
||||
("members", ("member_level_id",), "member_levels", ("id",)),
|
||||
("member_tag_links", ("memberTagId",), "memberTags", ("id",)),
|
||||
("visits", ("gate_id",), "gates", ("id",)),
|
||||
("visits", ("member_id",), "members", ("id",)),
|
||||
("stores", ("area_id",), "areas", ("id",)),
|
||||
("products", ("category_id",), "categories", ("id",)),
|
||||
("store_sales", ("productId",), "products", ("id",)),
|
||||
("store_sales", ("member_id",), "members", ("id",)),
|
||||
("inventory_snapshots", ("storeID",), "stores", ("id",)),
|
||||
("inventory_snapshots", ("productId",), "products", ("id",)),
|
||||
("coupons", ("campaign_id",), "campaigns", ("id",)),
|
||||
("reviews", ("visit_id",), "visits", ("id",)),
|
||||
("complaints", ("visit_id",), "visits", ("id",)),
|
||||
("event_bookings", ("event_id",), "events", ("id",)),
|
||||
("event_bookings", ("member_id",), "members", ("id",)),
|
||||
("weather_daily", ("park_id",), "parks", ("id",)),
|
||||
("ticket_inventory", ("ticket_type_id",), "ticket_types", ("id",)),
|
||||
("sms_logs", ("member_id",), "members", ("id",)),
|
||||
("purchase_order_lines", ("purchase_order_id",), "purchase_orders", ("id",)),
|
||||
("employees", ("store_id",), "stores", ("id",)),
|
||||
("tour_groups", ("agency_id",), "agencies", ("id",)),
|
||||
("tour_assignments", ("tour_group_id",), "tour_groups", ("id",)),
|
||||
("lost_and_found", ("park_id",), "parks", ("id",)),
|
||||
# 视图没有约束，按列名同样推得出来：按日汇总的入园人数照样能按 park_id 连到景区
|
||||
("v_daily_visits", ("park_id",), "parks", ("id",)),
|
||||
})
|
||||
|
||||
#: 看起来像关联、命名推断必须放过的列：(表, 列) → 原因
|
||||
NOT_INFERRED = {
|
||||
("employees", "employee_id"): "自引用：工号，不是指向本表",
|
||||
("tour_assignments", "guide_id"): "多个候选：guide 与 guides",
|
||||
("parking_records", "lot_id"): "没有名为 lot / lots 的表",
|
||||
("operation_logs", "operator_id"): "没有名为 operator / operators 的表",
|
||||
("categories", "parentId"): "没有名为 parent / parents 的表",
|
||||
("feedback_forms", "store_id"): "类型对不上：文字列指向整数主键",
|
||||
}
|
||||
|
||||
#: 唯一约束 / 唯一索引：表 → 列组
|
||||
UNIQUE = {
|
||||
"parks": [["park_code"]], "channels": [["channel_code"]], "orders": [["order_no"]],
|
||||
"members": [["member_no"]], "visits": [["ticket_no"]], "stores": [["store_code"]],
|
||||
"products": [["sku"]], "coupons": [["code"]], "dim_dates": [["biz_date"]],
|
||||
}
|
||||
|
||||
_PARKS = [(1, "P01", "青松谷景区", "甲市", "2015-05-01"), (2, "P02", "白鹭湾景区", "乙市", "2018-04-20"),
|
||||
(3, "P03", "星河乐园", "甲市", "2021-10-01")]
|
||||
_CHANNELS = [(1, "WEB", "官网", "线上直销"), (2, "MINI", "小程序", "线上直销"), (3, "BOX", "窗口售票", "线下"),
|
||||
(4, "AGT", "旅行社分销", "分销"), (5, "OTA1", "第三方平台甲", "分销"), (6, "OTA2", "第三方平台乙", "分销")]
|
||||
_LEVELS = [(1, "普通会员", 1.0), (2, "银卡会员", 0.95), (3, "金卡会员", 0.9), (4, "年卡会员", 0.0)]
|
||||
|
||||
|
||||
def _day(rng: random.Random) -> str:
|
||||
month = rng.choice((7, 8, 9))
|
||||
return f"2026-{month:02d}-{rng.randint(1, 30 if month == 9 else 31):02d}"
|
||||
|
||||
|
||||
def build(path: str | Path, *, seed: int = 20261003) -> Path:
|
||||
"""在 path 写一个新的景区业务库并返回路径。path 已存在时先删掉：同样的种子造出同样的内容。"""
|
||||
path = Path(path)
|
||||
if path.exists():
|
||||
path.unlink()
|
||||
rng = random.Random(seed)
|
||||
db = sqlite3.connect(path)
|
||||
try:
|
||||
db.executescript(DDL)
|
||||
db.executemany("INSERT INTO parks VALUES (?,?,?,?,?)", _PARKS)
|
||||
db.executemany("INSERT INTO areas VALUES (?,?,?)",
|
||||
[(i, 1 + (i - 1) // 2, f"{'东西南北中'[i % 5]}片区") for i in range(1, 7)])
|
||||
db.executemany("INSERT INTO gates VALUES (?,?,?,?)",
|
||||
[(i, 1 + (i - 1) % 3, f"G{i:02d}", f"{i} 号闸口") for i in range(1, 9)])
|
||||
db.executemany("INSERT INTO channels VALUES (?,?,?,?)", _CHANNELS)
|
||||
types = [(i, 1 + (i - 1) // 3, f"T{i:02d}", name, price)
|
||||
for i, (name, price) in enumerate([("成人票", 120.0), ("儿童票", 60.0), ("夜场票", 80.0)] * 3, 1)]
|
||||
db.executemany("INSERT INTO ticket_types VALUES (?,?,?,?,?)", types)
|
||||
db.executemany("INSERT INTO member_levels VALUES (?,?,?)", _LEVELS)
|
||||
db.executemany("INSERT INTO members VALUES (?,?,?,?,?,?)",
|
||||
[(i, f"M{i:05d}", f"会员{i}", rng.randint(1, 4), _day(rng), 1 if rng.random() > 0.05 else 0)
|
||||
for i in range(1, 201)])
|
||||
db.executemany('INSERT INTO "memberTags" VALUES (?,?)',
|
||||
[(i, t) for i, t in enumerate(["亲子", "夜游", "摄影", "年卡", "团队"], 1)])
|
||||
db.executemany("INSERT INTO member_tag_links VALUES (?,?,?)",
|
||||
[(i, rng.randint(1, 200), rng.randint(1, 5)) for i in range(1, 301)])
|
||||
orders, items = [], []
|
||||
for i in range(1, 601):
|
||||
qty = rng.randint(1, 4)
|
||||
tid = rng.randint(1, 9)
|
||||
amount = qty * types[tid - 1][4]
|
||||
orders.append((i, f"OD{i:06d}", rng.choice([None, rng.randint(1, 200)]), rng.randint(1, 6),
|
||||
f"{_day(rng)} {rng.randint(8, 20):02d}:{rng.randint(0, 59):02d}:00", amount,
|
||||
1 if rng.random() > 0.08 else 2))
|
||||
items.append((i, i, tid, qty, types[tid - 1][4], amount))
|
||||
db.executemany("INSERT INTO orders VALUES (?,?,?,?,?,?,?)", orders)
|
||||
db.executemany("INSERT INTO order_items VALUES (?,?,?,?,?,?)", items)
|
||||
visits = [(i, f"TK{i:07d}", 1 + i % 3, rng.randint(1, 8), rng.randint(1, 9), rng.choice([None, rng.randint(1, 200)]),
|
||||
f"{_day(rng)} {rng.randint(8, 21):02d}:{rng.randint(0, 59):02d}:00", rng.randint(1, 5),
|
||||
1 if rng.random() > 0.04 else 0) for i in range(1, 1501)]
|
||||
db.executemany("INSERT INTO visits VALUES (?,?,?,?,?,?,?,?,?)", visits)
|
||||
db.executemany("INSERT INTO channel_visits VALUES (?,?,?)", [(rng.randint(1, 6), i, 1.0) for i in range(1, 1501)])
|
||||
db.executemany("INSERT INTO stores VALUES (?,?,?,?,?,?)",
|
||||
[(i, f"S{i:03d}", f"{i} 号门店", 1 + i % 3, 1 + i % 6, "2020-01-01") for i in range(1, 11)])
|
||||
db.executemany("INSERT INTO categories VALUES (?,?,?)",
|
||||
[(1, "餐饮", None), (2, "文创", None), (3, "饮品", 1), (4, "小吃", 1), (5, "纪念品", 2)])
|
||||
db.executemany("INSERT INTO products VALUES (?,?,?,?,?)",
|
||||
[(i, f"SKU{i:04d}", f"商品{i}", rng.randint(1, 5), float(rng.randint(5, 200))) for i in range(1, 41)])
|
||||
db.executemany("INSERT INTO store_sales VALUES (?,?,?,?,?,?,?)",
|
||||
[(i, rng.randint(1, 10), rng.randint(1, 40), rng.choice([None, rng.randint(1, 200)]),
|
||||
f"{_day(rng)} 12:00:00", rng.randint(1, 3), float(rng.randint(5, 300))) for i in range(1, 801)])
|
||||
db.executemany("INSERT INTO dim_dates VALUES (?,?,?,?)",
|
||||
[(20260700 + d, f"2026-07-{d:02d}", d % 7, int(d % 7 in (5, 6))) for d in range(1, 32)])
|
||||
db.executemany("INSERT INTO holidays VALUES (?,?,?)", [("2026-10-01", "国庆节", 1), ("2026-10-02", "国庆节", 1)])
|
||||
db.executemany("INSERT INTO sys_config VALUES (?,?)", [("currency", "CNY"), ("timezone", "+08:00")])
|
||||
db.executemany("INSERT INTO guide VALUES (?,?)", [(1, "旧导游档案")])
|
||||
db.executemany("INSERT INTO guides VALUES (?,?,?)", [(1, "导游甲", 1), (2, "导游乙", 1)])
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
return path
|
||||
@@ -0,0 +1,199 @@
|
||||
"""数据目录接口：/api/datasources/{id}/catalog 下的表清单、单表、整份提交、单项审阅、起草。
|
||||
|
||||
所有写操作记录请求头 X-Actor 的署名；版本不符一律 409，detail 说明这张表刚被修改过、请重新载入。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from urllib.parse import quote
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from app.data import catalog
|
||||
from app.data.engine import engines
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource
|
||||
from app.main import app as _fastapi_app
|
||||
from tests.fixtures.catalog import scenic
|
||||
from tests.fixtures.catalog.fake_model import FakeModel
|
||||
|
||||
ACTOR = {"X-Actor": quote("王敏")}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
async with AsyncClient(transport=ASGITransport(app=_fastapi_app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def source_id(client, scenic_db):
|
||||
"""接上合成景区库并探查结构（走真实的探查接口）。"""
|
||||
async with SessionLocal() as session:
|
||||
row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=scenic_db)
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
sid = row.id
|
||||
r = await client.post(f"/api/datasources/{sid}/introspect")
|
||||
assert r.status_code == 200, r.text
|
||||
yield sid
|
||||
await engines.invalidate(sid)
|
||||
|
||||
|
||||
def _base(sid: str) -> str:
|
||||
return f"/api/datasources/{sid}/catalog"
|
||||
|
||||
|
||||
async def test_list_includes_tables_without_notes(client, source_id):
|
||||
r = await client.get(_base(source_id))
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
rows = {t["table_name"]: t for t in body["tables"]}
|
||||
assert len(rows) == scenic.TABLE_COUNT + len(scenic.VIEWS)
|
||||
visits = rows["visits"]
|
||||
assert visits["version"] == 0 and visits["label"] is None and visits["usage"] == 0
|
||||
assert visits["counts"] == {"proposed": 0, "verified": 0, "confirmed": 0, "rejected": 0}
|
||||
assert visits["relations"] == 0 and visits["updated_at"] is None
|
||||
assert body["system_notes"] is False
|
||||
|
||||
|
||||
async def test_draft_then_read_table(client, source_id):
|
||||
r = await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits", "nope"], "use_model": False},
|
||||
headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
by = {t["table_name"]: t for t in body["tables"]}
|
||||
# visits：两条外键（景区、票种）+ 两条命名推断（闸口、会员）
|
||||
assert by["visits"]["added"] == 4 and by["visits"]["version"] == 1 and by["visits"]["error"] is None
|
||||
assert by["nope"]["error"] and body["model_used"] is False
|
||||
assert body["total"] == {"added": 4, "updated": 0, "removed": 0}
|
||||
|
||||
r = await client.get(f"{_base(source_id)}/visits")
|
||||
assert r.status_code == 200
|
||||
detail = r.json()
|
||||
assert detail["version"] == 1 and detail["updated_by"] == "王敏"
|
||||
sources = sorted(rel["source"] for rel in detail["notes"]["relations"])
|
||||
assert sources == ["fk", "fk", "name", "name"]
|
||||
structure = detail["structure"]
|
||||
assert structure["primary_key"] == ["id"]
|
||||
assert {fk["to_table"] for fk in structure["foreign_keys"]} == {"parks", "ticket_types"}
|
||||
assert ["ticket_no"] in structure["unique"]
|
||||
assert any(c["name"] == "visit_time" for c in structure["columns"])
|
||||
|
||||
listed = {t["table_name"]: t for t in (await client.get(_base(source_id))).json()["tables"]}
|
||||
assert listed["visits"]["counts"] == {"proposed": 2, "verified": 2, "confirmed": 0, "rejected": 0}
|
||||
assert listed["visits"]["relations"] == 4 and listed["visits"]["version"] == 1
|
||||
|
||||
|
||||
async def test_put_marks_edits_human_confirmed_and_bumps_version(client, source_id):
|
||||
await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR)
|
||||
detail = (await client.get(f"{_base(source_id)}/visits")).json()
|
||||
notes = detail["notes"]
|
||||
notes["label"] = {"value": "入园记录"}
|
||||
notes["grain"] = {"value": "每张门票每次检票一行", "source": "llm", "status": "proposed"}
|
||||
r = await client.put(f"{_base(source_id)}/visits", json={"notes": notes, "if_version": detail["version"]},
|
||||
headers={"X-Actor": quote("李雷")})
|
||||
assert r.status_code == 200, r.text
|
||||
saved = r.json()
|
||||
assert saved["version"] == 2 and saved["updated_by"] == "李雷"
|
||||
assert saved["notes"]["label"]["source"] == "human" and saved["notes"]["label"]["status"] == "confirmed"
|
||||
assert saved["notes"]["grain"]["source"] == "human"
|
||||
assert {r["source"] for r in saved["notes"]["relations"]} == {"fk", "name"} # 没动的保持原样
|
||||
|
||||
|
||||
async def test_put_with_stale_version_conflicts(client, source_id):
|
||||
await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR)
|
||||
r = await client.put(f"{_base(source_id)}/visits", json={"notes": {"label": {"value": "x"}}, "if_version": 0},
|
||||
headers=ACTOR)
|
||||
assert r.status_code == 409
|
||||
assert "刚被修改过" in r.json()["detail"] and "重新载入" in r.json()["detail"]
|
||||
|
||||
|
||||
async def test_put_rejects_malformed_notes(client, source_id):
|
||||
r = await client.put(f"{_base(source_id)}/visits",
|
||||
json={"notes": {"formula": {"value": "sum(amount)"}}, "if_version": 0}, headers=ACTOR)
|
||||
assert r.status_code == 422
|
||||
assert "formula" in r.json()["detail"]
|
||||
|
||||
|
||||
async def test_review_confirm_reject_and_conflict(client, source_id):
|
||||
await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR)
|
||||
detail = (await client.get(f"{_base(source_id)}/visits")).json()
|
||||
gate = next(r for r in detail["notes"]["relations"] if r["columns"] == ["gate_id"])
|
||||
r = await client.post(f"{_base(source_id)}/visits/review",
|
||||
json={"path": f"relations.{gate['id']}", "action": "confirm", "if_version": 1},
|
||||
headers={"X-Actor": quote("韩梅")})
|
||||
assert r.status_code == 200, r.text
|
||||
after = r.json()
|
||||
assert after["version"] == 2 and after["updated_by"] == "韩梅"
|
||||
assert next(x for x in after["notes"]["relations"] if x["id"] == gate["id"])["status"] == "confirmed"
|
||||
stale = await client.post(f"{_base(source_id)}/visits/review",
|
||||
json={"path": f"relations.{gate['id']}", "action": "reject", "if_version": 1})
|
||||
assert stale.status_code == 409 and "重新载入" in stale.json()["detail"]
|
||||
bad = await client.post(f"{_base(source_id)}/visits/review",
|
||||
json={"path": "columns.visit_time.formula", "action": "confirm", "if_version": 2})
|
||||
assert bad.status_code == 422
|
||||
wrong = await client.post(f"{_base(source_id)}/visits/review",
|
||||
json={"path": "grain", "action": "approve", "if_version": 2})
|
||||
assert wrong.status_code == 422
|
||||
|
||||
|
||||
async def test_unknown_table_and_source(client, source_id):
|
||||
assert (await client.get(f"{_base(source_id)}/nope")).status_code == 404
|
||||
assert (await client.get(_base("missing-source"))).status_code == 404
|
||||
r = await client.put(f"{_base(source_id)}/nope", json={"notes": {}, "if_version": 0})
|
||||
assert r.status_code == 404
|
||||
|
||||
|
||||
async def test_draft_defaults_to_top_tables_and_model_via_settings(client, source_id, monkeypatch):
|
||||
model = FakeModel({"visits": {"label": "入园记录", "kind": "fact"}})
|
||||
|
||||
async def fake_resolve(session):
|
||||
return model, "fake-model"
|
||||
|
||||
async def usage(session, source):
|
||||
return {"visits": 7}
|
||||
|
||||
monkeypatch.setattr(catalog, "resolve_draft_model", fake_resolve)
|
||||
monkeypatch.setattr(catalog, "table_usage", usage)
|
||||
r = await client.post(f"{_base(source_id)}/draft", json={"use_model": True}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
assert len(body["tables"]) == 20 and body["tables"][0]["table_name"] == "visits"
|
||||
assert body["model_used"] is True and body["model"] == "fake-model"
|
||||
detail = (await client.get(f"{_base(source_id)}/visits")).json()
|
||||
assert detail["notes"]["label"]["source"] == "llm" and detail["usage"] == 7
|
||||
|
||||
|
||||
async def test_draft_reports_model_unavailable(client, source_id, monkeypatch):
|
||||
async def unavailable(session):
|
||||
raise catalog.CatalogModelUnavailable("未配置模型接入，无法由助手起草数据目录。请到「设置 → 模型接入」添加")
|
||||
|
||||
monkeypatch.setattr(catalog, "resolve_draft_model", unavailable)
|
||||
r = await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"], "use_model": True})
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["model_used"] is False and "模型接入" in body["model_error"]
|
||||
assert body["tables"][0]["added"] == 4 # 外键和命名推断照常写入
|
||||
|
||||
|
||||
async def test_draft_without_schema_is_refused(client):
|
||||
async with SessionLocal() as session:
|
||||
row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=":memory:")
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
sid = row.id
|
||||
r = await client.post(f"{_base(sid)}/draft", json={})
|
||||
assert r.status_code == 409 and "探查结构" in r.json()["detail"]
|
||||
assert (await client.get(_base(sid))).json()["tables"] == []
|
||||
|
||||
|
||||
async def test_schema_endpoint_detail_carries_catalog(client, source_id):
|
||||
"""数据源页「表结构」里给模型看的那段文本和 db_schema 工具同一个说法，也附上目录。"""
|
||||
before = (await client.get(f"/api/datasources/{source_id}/schema", params={"table": "visits"})).json()
|
||||
assert catalog.NOTES_HEADING not in before["detail"]
|
||||
await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR)
|
||||
after = (await client.get(f"/api/datasources/{source_id}/schema", params={"table": "visits"})).json()
|
||||
assert catalog.NOTES_HEADING in after["detail"] and "gate_id → gates.id" in after["detail"]
|
||||
assert after["detail"].startswith(before["detail"])
|
||||
@@ -0,0 +1,375 @@
|
||||
"""数据目录的起草：数据库注释、外键约束、命名推断、模型起草，以及并入已有目录。
|
||||
|
||||
这一阶段起草不发数据库查询（数据剖析归阶段 2），只看探查缓存里的结构。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from app.data import catalog
|
||||
from app.data.catalog import make_item
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource
|
||||
from tests.fixtures.catalog.fake_model import FakeModel
|
||||
|
||||
|
||||
def _col(name: str, type_: str = "INTEGER", comment: str | None = None, nullable: bool = True) -> dict:
|
||||
return {"name": name, "type": type_, "nullable": nullable, "comment": comment}
|
||||
|
||||
|
||||
def _table(*cols: dict, pk=("id",), fks=None, unique=None, comment=None) -> dict:
|
||||
meta = {"qualified": "", "schema": None, "columns": list(cols), "primary_key": list(pk), "comment": comment,
|
||||
"is_view": False}
|
||||
if fks is not None:
|
||||
meta["foreign_keys"] = fks
|
||||
if unique is not None:
|
||||
meta["unique"] = unique
|
||||
return meta
|
||||
|
||||
|
||||
def _source(tables: dict, **kw) -> SimpleNamespace:
|
||||
for name, meta in tables.items():
|
||||
meta["qualified"] = name
|
||||
return SimpleNamespace(id=kw.pop("id", "src"), name="scenic", kind="sqlite", origin=kw.pop("origin", "manual"),
|
||||
schema_cache={"tables": tables, **kw})
|
||||
|
||||
|
||||
def _rels(notes: dict) -> set[tuple]:
|
||||
return {(tuple(r["columns"]), r["to_table"], tuple(r["to_columns"]), r["source"], r["status"])
|
||||
for r in notes.get("relations", [])}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 数据库注释
|
||||
|
||||
|
||||
def test_comments_become_labels_or_meanings():
|
||||
src = _source({"visits": _table(
|
||||
_col("id"), _col("amount", "REAL", comment="实收金额"),
|
||||
_col("status", comment="状态：1 表示有效，0 表示作废；作废记录不计入客流"),
|
||||
comment="入园记录")})
|
||||
draft = catalog.draft_structure(src, "visits")
|
||||
assert draft.notes["label"] == make_item("入园记录", "comment")
|
||||
assert draft.notes["columns"]["amount"]["label"] == make_item("实收金额", "comment")
|
||||
assert draft.notes["columns"]["status"]["meaning"]["value"].startswith("状态：1 表示有效")
|
||||
assert "label" not in draft.notes["columns"]["status"]
|
||||
assert "comment" in draft.covered
|
||||
|
||||
|
||||
def test_long_table_comment_goes_to_description():
|
||||
src = _source({"visits": _table(_col("id"), comment="游客每次检票入园记一行，团队票按人数拆成多行")})
|
||||
draft = catalog.draft_structure(src, "visits")
|
||||
assert "label" not in draft.notes
|
||||
assert draft.notes["description"]["value"].startswith("游客每次检票入园")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kw", [{"origin": "upload"}, {"import_mode": "recipe"}])
|
||||
def test_imported_table_comments_are_not_copied(kw):
|
||||
"""导入表格的说明是按核对结果生成的（recipe_notes），数据目录不重复它们。"""
|
||||
src = _source({"日报": _table(_col("日期", "TEXT", comment="格式 YYYY-MM-DD"), _col("客流", comment="单位：人次"),
|
||||
pk=(), comment="粒度：每个日期")}, **kw)
|
||||
assert catalog.system_notes_source(src)
|
||||
draft = catalog.draft_structure(src, "日报")
|
||||
assert "label" not in draft.notes and "description" not in draft.notes
|
||||
assert "columns" not in draft.notes
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 外键约束
|
||||
|
||||
|
||||
def test_foreign_keys_become_verified_relations():
|
||||
src = _source({
|
||||
"visits": _table(_col("id"), _col("park_id"), fks=[
|
||||
{"columns": ["park_id"], "to_table": "parks", "to_columns": ["id"]}]),
|
||||
"parks": _table(_col("id"), fks=[]),
|
||||
})
|
||||
draft = catalog.draft_structure(src, "visits")
|
||||
assert _rels(draft.notes) == {(("park_id",), "parks", ("id",), "fk", "verified")}
|
||||
rel = draft.notes["relations"][0]
|
||||
assert rel["cardinality"] == "many_to_one"
|
||||
assert rel["id"] == catalog.relation_id("visits", ["park_id"], "parks", ["id"])
|
||||
assert "fk" in draft.covered
|
||||
|
||||
|
||||
def test_foreign_key_without_target_columns_points_at_primary_key_and_one_to_one():
|
||||
src = _source({
|
||||
"member_profiles": _table(_col("member_id"), _col("nickname", "TEXT"), pk=("member_id",),
|
||||
fks=[{"columns": ["member_id"], "to_table": "members", "to_columns": []}]),
|
||||
"members": _table(_col("id"), fks=[]),
|
||||
})
|
||||
rel = catalog.draft_structure(src, "member_profiles").notes["relations"][0]
|
||||
assert rel["to_columns"] == ["id"] and rel["cardinality"] == "one_to_one"
|
||||
|
||||
|
||||
def test_old_cache_without_foreign_keys_does_not_cover_fk():
|
||||
"""旧缓存里没有 foreign_keys：不知道有没有外键，不能据此删掉已有的外键关系。"""
|
||||
src = _source({"visits": _table(_col("id"), _col("park_id")), "parks": _table(_col("id"))})
|
||||
draft = catalog.draft_structure(src, "visits")
|
||||
assert "fk" not in draft.covered
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 命名推断
|
||||
|
||||
|
||||
def _infer(tables: dict, table: str) -> set[tuple]:
|
||||
src = _source(tables)
|
||||
return {(tuple(r["columns"]), r["to_table"], tuple(r["to_columns"]))
|
||||
for r in catalog.infer_name_relations(src.schema_cache, table)}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("column, target", [
|
||||
("channel_id", "channels"), # 复数
|
||||
("category_id", "categories"), # y → ies
|
||||
("storeId", "stores"), # 驼峰
|
||||
("storeID", "stores"), # xxxID
|
||||
("ticket_type_id", "ticket_types"), # 多段下划线
|
||||
("memberTagId", "memberTags"), # 驼峰表名
|
||||
("member_tag_id", "memberTags"), # 下划线列名对驼峰表名
|
||||
("park_id", "park"), # 单数表名
|
||||
("CHANNEL_ID", "CHANNELS"), # 大写（Oracle）
|
||||
])
|
||||
def test_name_inference_positive(column, target):
|
||||
tables = {"facts": _table(_col("id"), _col(column)), target: _table(_col("id"))}
|
||||
pk = "id"
|
||||
assert _infer(tables, "facts") == {((column,), target, (pk,))}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("tables, why", [
|
||||
({"employees": _table(_col("id"), _col("employee_id", "TEXT"))}, "自引用"),
|
||||
({"facts": _table(_col("id"), _col("guide_id")), "guide": _table(_col("id")), "guides": _table(_col("id"))},
|
||||
"多个候选"),
|
||||
({"facts": _table(_col("id"), _col("lot_id")), "parking_lots": _table(_col("id"))}, "没有对应的表"),
|
||||
({"facts": _table(_col("id"), _col("store_id", "VARCHAR(20)")), "stores": _table(_col("id"))}, "类型对不上"),
|
||||
({"facts": _table(_col("id"), _col("day_id")), "days": _table(_col("park_id"), _col("d"), pk=("park_id", "d"))},
|
||||
"被指向的表是复合主键"),
|
||||
({"facts": _table(_col("id"), _col("tag_id")), "tags": _table(_col("name", "TEXT"), pk=())}, "被指向的表没有主键"),
|
||||
({"facts": _table(_col("id"), _col("x_id")), "xs": _table(_col("id"))}, "词根太短"),
|
||||
({"facts": _table(_col("id"), _col("paid")), "pas": _table(_col("id"))}, "不是 _id 结尾"),
|
||||
({"facts": _table(_col("id"), _col("store_id"), fks=[{"columns": ["store_id"], "to_table": "shops",
|
||||
"to_columns": ["id"]}]),
|
||||
"stores": _table(_col("id")), "shops": _table(_col("id"))}, "已有外键约束的列"),
|
||||
])
|
||||
def test_name_inference_negative(tables, why):
|
||||
assert _infer(tables, next(iter(tables))) == set(), why
|
||||
|
||||
|
||||
def test_name_inference_one_to_one_when_column_is_own_key():
|
||||
tables = {"member_profiles": _table(_col("member_id"), pk=("member_id",)), "members": _table(_col("id"))}
|
||||
rel = catalog.infer_name_relations(_source(tables).schema_cache, "member_profiles")[0]
|
||||
assert rel["cardinality"] == "one_to_one" and rel["source"] == "name" and rel["status"] == "proposed"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 模型起草
|
||||
|
||||
|
||||
_VISITS_REPLY = {
|
||||
"label": "入园记录", "grain": "每张门票每次检票一行", "kind": "fact", "keys": ["ticket_no"],
|
||||
"business_date": {"column": "visit_time", "rule": "按检票时间计", "timezone": "+08:00"},
|
||||
"columns": [
|
||||
{"name": "visitor_count", "label": "入园人数", "meaning": "这次检票进园的人数", "unit": "人", "measure": "flow"},
|
||||
{"name": "status", "label": "状态", "measure": "status"},
|
||||
{"name": "ghost", "label": "不存在的列"}, # 丢掉
|
||||
{"name": "ticket_no", "measure": "money"}, # 度量类型不在取值里：丢掉这一项
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _scenic_tables() -> dict:
|
||||
return {
|
||||
"visits": _table(_col("id"), _col("ticket_no", "TEXT"), _col("visit_time", "TEXT"), _col("visitor_count"),
|
||||
_col("status"), fks=[]),
|
||||
"channels": _table(_col("id"), _col("name", "TEXT"), fks=[]),
|
||||
}
|
||||
|
||||
|
||||
async def test_model_draft_maps_fields_and_drops_invalid_values():
|
||||
src = _source(_scenic_tables())
|
||||
model = FakeModel({"visits": _VISITS_REPLY, "channels": {"label": "渠道", "kind": "planet",
|
||||
"keys": ["nope"]}})
|
||||
got = await catalog.draft_with_model(model, src, ["visits", "channels"])
|
||||
visits = got["visits"].notes
|
||||
assert visits["label"] == make_item("入园记录", "llm")
|
||||
assert visits["grain"]["value"] == "每张门票每次检票一行" and visits["kind"]["value"] == "fact"
|
||||
assert visits["keys"]["value"] == ["ticket_no"]
|
||||
assert visits["business_date"]["value"] == {"column": "visit_time", "rule": "按检票时间计", "timezone": "+08:00"}
|
||||
assert visits["columns"]["visitor_count"]["unit"] == make_item("人", "llm")
|
||||
assert visits["columns"]["status"]["measure"]["value"] == "status"
|
||||
assert "ghost" not in visits["columns"] and "ticket_no" not in visits["columns"]
|
||||
channels = got["channels"].notes
|
||||
assert channels == {"label": make_item("渠道", "llm")} # 表类型、主键不合规的都丢掉
|
||||
assert got["visits"].covered == frozenset({"llm"})
|
||||
assert catalog.validate_notes(visits) == []
|
||||
|
||||
|
||||
async def test_model_prompt_carries_structure_and_no_data():
|
||||
src = _source(_scenic_tables())
|
||||
model = FakeModel({"visits": _VISITS_REPLY})
|
||||
await catalog.draft_with_model(model, src, ["visits"])
|
||||
prompt = model.calls[0]
|
||||
assert "表 visits" in prompt and "ticket_no" in prompt and "visit_time" in prompt
|
||||
|
||||
|
||||
async def test_model_failure_only_affects_its_batch():
|
||||
tables = {f"t{i:02d}": _table(_col("id"), *[_col(f"c{j}") for j in range(60)]) for i in range(4)}
|
||||
src = _source(tables)
|
||||
model = FakeModel({name: {"label": f"表{name}"} for name in tables}, fail={"t00"})
|
||||
got = await catalog.draft_with_model(model, src, list(tables))
|
||||
assert isinstance(got["t00"], str) and "网关超时" in got["t00"]
|
||||
assert all(got[n].notes["label"]["value"] == f"表{n}" for n in ("t01", "t02", "t03"))
|
||||
assert len(model.calls) >= 2 # 按表分批，不是一次全发
|
||||
|
||||
|
||||
async def test_model_without_structured_output_falls_back_to_text():
|
||||
"""有的网关不支持工具调用：结构化输出一调就报错，退回纯文本、从正文里取 JSON。"""
|
||||
from langchain_core.messages import AIMessage
|
||||
|
||||
class TextOnly:
|
||||
def with_structured_output(self, schema, **_):
|
||||
raise NotImplementedError
|
||||
|
||||
async def ainvoke(self, messages):
|
||||
return AIMessage(content='草稿如下：{"tables": [{"table": "visits", "label": "入园记录"}]}')
|
||||
|
||||
got = await catalog.draft_with_model(TextOnly(), _source(_scenic_tables()), ["visits"])
|
||||
assert got["visits"].notes == {"label": make_item("入园记录", "llm")}
|
||||
|
||||
|
||||
async def test_model_draft_respects_system_notes_of_imported_tables():
|
||||
"""导入表格：表说明已经写了粒度、列说明写了单位，模型起草的粒度和单位不进目录，免得重复或矛盾。"""
|
||||
src = _source({"日报": _table(_col("日期", "TEXT", comment="格式 YYYY-MM-DD"), _col("客流"), pk=(),
|
||||
comment="粒度：每个日期")}, origin="upload")
|
||||
model = FakeModel({"日报": {"label": "客流日报", "grain": "每天一行", "columns": [
|
||||
{"name": "日期", "meaning": "统计日期", "unit": "天", "measure": "attribute"},
|
||||
{"name": "客流", "meaning": "当天入园人次", "unit": "人次", "measure": "flow"}]}})
|
||||
got = (await catalog.draft_with_model(model, src, ["日报"]))["日报"].notes
|
||||
assert "grain" not in got and got["label"]["value"] == "客流日报"
|
||||
assert "meaning" not in got["columns"]["日期"] and "unit" not in got["columns"]["日期"]
|
||||
assert got["columns"]["日期"]["measure"]["value"] == "attribute"
|
||||
assert got["columns"]["客流"]["unit"]["value"] == "人次" # 这一列没有系统说明，照常起草
|
||||
assert "粒度：每个日期" in model.calls[0] # 系统说明交给模型作背景
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 起草并写入
|
||||
|
||||
|
||||
async def _db_source(tables: dict, **kw) -> DataSource:
|
||||
for name, meta in tables.items():
|
||||
meta["qualified"] = name
|
||||
async with SessionLocal() as session:
|
||||
row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=":memory:",
|
||||
schema_cache={"tables": tables}, **kw)
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
return row
|
||||
|
||||
|
||||
async def test_draft_catalog_writes_and_is_idempotent():
|
||||
tables = _scenic_tables()
|
||||
tables["visits"]["foreign_keys"] = [{"columns": ["channel_id"], "to_table": "channels", "to_columns": ["id"]}]
|
||||
tables["visits"]["columns"].append(_col("channel_id"))
|
||||
tables["visits"]["columns"].append(_col("gate_id"))
|
||||
tables["gates"] = _table(_col("id"), fks=[])
|
||||
row = await _db_source(tables)
|
||||
model = FakeModel({"visits": _VISITS_REPLY})
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=model,
|
||||
actor="王敏")
|
||||
result = report.tables[0]
|
||||
assert result.table == "visits" and result.error is None and result.version == 1
|
||||
# 表级 5 项（中文名、粒度、表类型、业务主键、业务日期）+ 列级 6 项 + 外键和命名推断各一条关系
|
||||
assert (result.added, result.updated, result.removed) == (13, 0, 0)
|
||||
assert report.model_used and report.model_error is None
|
||||
async with SessionLocal() as session:
|
||||
entry = await catalog.read_entry(session, row.id, "visits")
|
||||
assert entry.updated_by == "王敏"
|
||||
assert _rels(entry.notes) == {(("channel_id",), "channels", ("id",), "fk", "verified"),
|
||||
(("gate_id",), "gates", ("id",), "name", "proposed")}
|
||||
again = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=model, actor=None)
|
||||
assert (again.tables[0].added, again.tables[0].updated, again.tables[0].removed) == (0, 0, 0)
|
||||
assert again.tables[0].version == 1 # 没有变化不升版本
|
||||
|
||||
|
||||
async def test_redraft_keeps_confirmed_and_rejected_items():
|
||||
tables = _scenic_tables()
|
||||
tables["visits"]["columns"].append(_col("gate_id"))
|
||||
tables["gates"] = _table(_col("id"), fks=[])
|
||||
row = await _db_source(tables)
|
||||
model = FakeModel({"visits": _VISITS_REPLY})
|
||||
async with SessionLocal() as session:
|
||||
await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=model, actor=None)
|
||||
entry = await catalog.read_entry(session, row.id, "visits")
|
||||
rid = entry.notes["relations"][0]["id"]
|
||||
entry = await catalog.review_entry(session, row.id, "visits", "grain", "confirm", if_version=entry.version,
|
||||
actor="李雷")
|
||||
entry = await catalog.review_entry(session, row.id, "visits", f"relations.{rid}", "reject",
|
||||
if_version=entry.version, actor="李雷")
|
||||
changed = FakeModel({"visits": {**_VISITS_REPLY, "grain": "每次入园一行", "label": "游客入园"}})
|
||||
report = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, model=changed,
|
||||
actor=None)
|
||||
after = await catalog.read_entry(session, row.id, "visits")
|
||||
assert after.notes["grain"]["value"] == "每张门票每次检票一行" and after.notes["grain"]["status"] == "confirmed"
|
||||
assert after.notes["relations"][0]["status"] == "rejected" # 驳回的关系留着，下次起草不再提出
|
||||
assert after.notes["label"]["value"] == "游客入园" # 同一来源、没确认过的项照常更新
|
||||
assert report.tables[0].updated == 1
|
||||
|
||||
|
||||
async def test_draft_catalog_model_unavailable_still_writes_other_sources(monkeypatch):
|
||||
tables = _scenic_tables()
|
||||
tables["visits"]["comment"] = "入园记录"
|
||||
row = await _db_source(tables)
|
||||
|
||||
async def unavailable(session):
|
||||
raise catalog.CatalogModelUnavailable("未配置模型接入，无法起草。请到「设置 → 模型接入」添加")
|
||||
|
||||
monkeypatch.setattr(catalog, "resolve_draft_model", unavailable)
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, tables=["visits"], use_model=True, actor=None)
|
||||
entry = await catalog.read_entry(session, row.id, "visits")
|
||||
assert not report.model_used and "模型接入" in report.model_error
|
||||
label = entry.notes["label"]
|
||||
assert label["updated_at"] and {k: v for k, v in label.items() if k != "updated_at"} == make_item("入园记录",
|
||||
"comment")
|
||||
|
||||
|
||||
async def test_draft_catalog_model_failure_on_one_table(monkeypatch):
|
||||
row = await _db_source(_scenic_tables())
|
||||
model = FakeModel({"visits": _VISITS_REPLY, "channels": {"label": "渠道"}}, fail={"visits"})
|
||||
monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1)
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, tables=["visits", "channels"], use_model=True,
|
||||
model=model, actor=None)
|
||||
by = {r.table: r for r in report.tables}
|
||||
assert by["visits"].model_error and "网关超时" in by["visits"].model_error
|
||||
assert by["channels"].model_error is None and by["channels"].added == 1
|
||||
|
||||
|
||||
async def test_draft_catalog_defaults_to_most_used_tables(monkeypatch):
|
||||
tables = {f"t{i:02d}": _table(_col("id"), fks=[]) for i in range(25)}
|
||||
row = await _db_source(tables)
|
||||
|
||||
async def usage(session, source):
|
||||
return {"t24": 9, "t03": 5, "t10": 5}
|
||||
|
||||
monkeypatch.setattr(catalog, "table_usage", usage)
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, use_model=False, actor=None)
|
||||
picked = [r.table for r in report.tables]
|
||||
assert len(picked) == catalog.DRAFT_DEFAULT_TABLES == 20
|
||||
assert picked[:3] == ["t24", "t03", "t10"] # 次数相同按表结构里的顺序
|
||||
assert picked[3:6] == ["t00", "t01", "t02"]
|
||||
|
||||
|
||||
async def test_draft_catalog_unknown_table_is_reported():
|
||||
row = await _db_source(_scenic_tables())
|
||||
async with SessionLocal() as session:
|
||||
report = await catalog.draft_catalog(session, row, tables=["nope"], use_model=False, actor=None)
|
||||
assert report.tables[0].error and "nope" in report.tables[0].error
|
||||
|
||||
|
||||
def test_draft_prompt_constants_are_model_only():
|
||||
"""提示词常量的名字带 _SYSTEM / _PROMPT：文案检查按名字把它们排除在界面文案之外。"""
|
||||
assert catalog.CATALOG_DRAFT_SYSTEM and catalog.CATALOG_DRAFT_PROMPT
|
||||
json.dumps(catalog.CATALOG_DRAFT_PROMPT_SCHEMA, ensure_ascii=False)
|
||||
@@ -0,0 +1,183 @@
|
||||
"""数据目录的合并、单项审阅和人工编辑规则（纯函数）。
|
||||
|
||||
- 起草结果并入已有目录：人工确认过、驳回过的项一律不动；同一来源的项可以更新；不同来源按可信程度
|
||||
（人工 > 外键 > 数据剖析 > 数据库注释 > 模型起草、命名推断）决定谁留下；这一轮完整算过的来源，
|
||||
已经没有的旧项删掉（外键约束被删了，原来那条「有确证」的关系不能留着）。
|
||||
- 单项审阅：confirm / reject / reset，路径写法 grain、columns.amount.measure、relations.<编号>。
|
||||
- 人工提交整份目录：改动过的项记为 human / confirmed，没动过的项保持原来的来源和状态。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from app.data import catalog
|
||||
from app.data.catalog import make_item
|
||||
|
||||
AT = "2026-10-03T08:00:00+00:00"
|
||||
|
||||
|
||||
def _rel(columns, to_table, to_columns, source="name", status=None, table="visits", **kw):
|
||||
return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns,
|
||||
"to_table": to_table, "to_columns": to_columns, "cardinality": "many_to_one", "coverage": None,
|
||||
"source": source, "status": status or catalog.initial_status(source), **kw}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 合并
|
||||
|
||||
|
||||
def test_merge_adds_new_items_and_counts():
|
||||
draft = {"label": make_item("入园记录", "llm"),
|
||||
"columns": {"amount": {"unit": make_item("元", "llm")}},
|
||||
"relations": [_rel(["gate_id"], "gates", ["id"])]}
|
||||
merged, stats = catalog.merge_notes({}, draft, at=AT)
|
||||
assert stats == catalog.MergeStats(added=3, updated=0, removed=0)
|
||||
assert merged["label"]["value"] == "入园记录" and merged["label"]["updated_at"] == AT
|
||||
assert merged["columns"]["amount"]["unit"]["status"] == "proposed"
|
||||
assert merged["relations"][0]["source"] == "name"
|
||||
|
||||
|
||||
def test_merge_never_overwrites_confirmed_or_rejected():
|
||||
existing = {"label": make_item("入园明细", "human"),
|
||||
"grain": make_item("每张票一行", "llm", "rejected"),
|
||||
"relations": [_rel(["member_id"], "members", ["id"], status="rejected")]}
|
||||
draft = {"label": make_item("入园记录", "llm"), "grain": make_item("每张票一行", "llm"),
|
||||
"relations": [_rel(["member_id"], "members", ["id"])]}
|
||||
merged, stats = catalog.merge_notes(existing, draft, covered={"llm", "name"}, at=AT)
|
||||
assert merged == existing
|
||||
assert stats == catalog.MergeStats(0, 0, 0)
|
||||
|
||||
|
||||
def test_merge_same_source_updates_value():
|
||||
existing = {"grain": make_item("每次入园一行", "llm", at="2026-09-01T00:00:00+00:00")}
|
||||
merged, stats = catalog.merge_notes(existing, {"grain": make_item("每张门票一次检票一行", "llm")}, at=AT)
|
||||
assert merged["grain"]["value"] == "每张门票一次检票一行" and merged["grain"]["updated_at"] == AT
|
||||
assert stats.updated == 1
|
||||
|
||||
|
||||
def test_merge_same_value_is_not_an_update():
|
||||
existing = {"grain": make_item("每次入园一行", "llm", at="2026-09-01T00:00:00+00:00")}
|
||||
merged, stats = catalog.merge_notes(existing, {"grain": make_item("每次入园一行", "llm")}, at=AT)
|
||||
assert merged == existing and stats == catalog.MergeStats(0, 0, 0)
|
||||
|
||||
|
||||
def test_merge_higher_rank_source_replaces_lower_but_not_the_reverse():
|
||||
# 外键约束（有确证）顶掉同一条关系的命名推断
|
||||
existing = {"relations": [_rel(["member_id"], "members", ["id"], source="name")]}
|
||||
merged, stats = catalog.merge_notes(existing, {"relations": [_rel(["member_id"], "members", ["id"], source="fk")]},
|
||||
at=AT)
|
||||
assert merged["relations"][0]["source"] == "fk" and merged["relations"][0]["status"] == "verified"
|
||||
assert stats.updated == 1
|
||||
# 数据库注释比模型起草可信：模型起草不顶掉注释，注释可以顶掉模型起草
|
||||
existing = {"label": make_item("入园记录", "comment")}
|
||||
merged, _ = catalog.merge_notes(existing, {"label": make_item("游客入园", "llm")}, at=AT)
|
||||
assert merged["label"]["source"] == "comment"
|
||||
merged, _ = catalog.merge_notes({"label": make_item("游客入园", "llm")}, {"label": make_item("入园记录", "comment")},
|
||||
at=AT)
|
||||
assert merged["label"] == {**make_item("入园记录", "comment"), "updated_at": AT}
|
||||
|
||||
|
||||
def test_merge_removes_stale_items_only_for_covered_sources():
|
||||
stale_fk = _rel(["store_id"], "stores", ["id"], source="fk")
|
||||
keep_name = _rel(["gate_id"], "gates", ["id"], source="name")
|
||||
existing = {"relations": [stale_fk, keep_name], "label": make_item("入园记录", "llm")}
|
||||
# 这一轮完整读过外键约束（covered 含 fk）而那条外键已经没了：删掉；命名推断和模型起草这一轮没算，留着
|
||||
merged, stats = catalog.merge_notes(existing, {}, covered={"fk"}, at=AT)
|
||||
assert merged["relations"] == [keep_name] and merged["label"]["value"] == "入园记录"
|
||||
assert stats.removed == 1
|
||||
|
||||
|
||||
def test_merge_keeps_column_entries_tidy():
|
||||
existing = {"columns": {"amount": {"unit": make_item("元", "llm")}}}
|
||||
merged, stats = catalog.merge_notes(existing, {}, covered={"llm"}, at=AT)
|
||||
assert merged == {} and stats.removed == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 单项审阅
|
||||
|
||||
|
||||
def _notes():
|
||||
return {"grain": make_item("每次入园一行", "llm"),
|
||||
"columns": {"amount": {"measure": make_item("flow", "llm")}},
|
||||
"relations": [_rel(["park_id"], "parks", ["id"], source="fk"), _rel(["gate_id"], "gates", ["id"])]}
|
||||
|
||||
|
||||
def test_review_confirm_reject_reset_on_each_kind_of_path():
|
||||
notes = _notes()
|
||||
rid = notes["relations"][1]["id"]
|
||||
out = catalog.review_item(notes, "grain", "confirm", at=AT)
|
||||
assert out["grain"]["status"] == "confirmed" and out["grain"]["source"] == "llm" and out["grain"]["updated_at"] == AT
|
||||
out = catalog.review_item(out, "columns.amount.measure", "reject", at=AT)
|
||||
assert out["columns"]["amount"]["measure"]["status"] == "rejected"
|
||||
out = catalog.review_item(out, f"relations.{rid}", "confirm", at=AT)
|
||||
assert out["relations"][1]["status"] == "confirmed"
|
||||
# reset 回到来源的初始状态：外键是 verified，其余是 proposed
|
||||
fk_id = notes["relations"][0]["id"]
|
||||
out = catalog.review_item(catalog.review_item(out, f"relations.{fk_id}", "reject", at=AT),
|
||||
f"relations.{fk_id}", "reset", at=AT)
|
||||
assert out["relations"][0]["status"] == "verified"
|
||||
out = catalog.review_item(out, "columns.amount.measure", "reset", at=AT)
|
||||
assert out["columns"]["amount"]["measure"]["status"] == "proposed"
|
||||
assert notes["grain"]["status"] == "proposed" # 不改入参
|
||||
|
||||
|
||||
def test_review_reset_of_human_item_removes_it():
|
||||
notes = {"label": make_item("入园记录", "human")}
|
||||
assert catalog.review_item(notes, "label", "reset", at=AT) == {}
|
||||
|
||||
|
||||
def test_review_column_names_may_contain_dots():
|
||||
notes = {"columns": {"a.b": {"unit": make_item("元", "llm")}}}
|
||||
out = catalog.review_item(notes, "columns.a.b.unit", "confirm", at=AT)
|
||||
assert out["columns"]["a.b"]["unit"]["status"] == "confirmed"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", ["", "nope", "grain.extra", "columns.amount", "columns.amount.formula",
|
||||
"columns.missing.measure", "relations.r000000000000", "label"])
|
||||
def test_review_bad_paths(path):
|
||||
with pytest.raises(catalog.CatalogPathError):
|
||||
catalog.review_item(_notes(), path, "confirm", at=AT)
|
||||
|
||||
|
||||
def test_review_bad_action():
|
||||
with pytest.raises(catalog.CatalogPathError):
|
||||
catalog.review_item(_notes(), "grain", "approve", at=AT)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 人工编辑
|
||||
|
||||
|
||||
def test_human_edit_marks_changed_items_as_human_confirmed_and_keeps_the_rest():
|
||||
existing = _notes()
|
||||
submitted = {
|
||||
"grain": {**existing["grain"], "value": "每张门票每次检票一行"}, # 改了值
|
||||
"columns": {"amount": {"measure": existing["columns"]["amount"]["measure"], # 没动
|
||||
"unit": {"value": "元"}}}, # 新填，没写来源和状态
|
||||
"relations": existing["relations"][:1], # 删掉一条关系
|
||||
"dedup": {"value": "同一票号只计一次", "source": "llm", "status": "verified"}, # 冒充别的来源、状态
|
||||
}
|
||||
out = catalog.apply_human_edit(existing, submitted, table="visits", at=AT)
|
||||
assert out["grain"] == {"value": "每张门票每次检票一行", "source": "human", "status": "confirmed", "updated_at": AT}
|
||||
assert out["columns"]["amount"]["measure"] == existing["columns"]["amount"]["measure"]
|
||||
assert out["columns"]["amount"]["unit"] == {"value": "元", "source": "human", "status": "confirmed",
|
||||
"updated_at": AT}
|
||||
assert out["dedup"]["source"] == "human" and out["dedup"]["status"] == "confirmed"
|
||||
assert len(out["relations"]) == 1 and out["relations"][0] == existing["relations"][0]
|
||||
assert catalog.validate_notes(out) == []
|
||||
|
||||
|
||||
def test_human_edit_status_only_change_keeps_source():
|
||||
existing = _notes()
|
||||
submitted = {**existing, "grain": {**existing["grain"], "status": "confirmed"}}
|
||||
out = catalog.apply_human_edit(existing, submitted, table="visits", at=AT)
|
||||
assert out["grain"]["source"] == "llm" and out["grain"]["status"] == "confirmed"
|
||||
# 不能把推断改成「有确证」：verified 只来自外键约束和数据剖析
|
||||
submitted = {**existing, "grain": {**existing["grain"], "status": "verified"}}
|
||||
assert catalog.apply_human_edit(existing, submitted, table="visits", at=AT)["grain"]["status"] == "proposed"
|
||||
|
||||
|
||||
def test_human_edit_relation_id_follows_its_ends():
|
||||
existing = _notes()
|
||||
moved = {**existing["relations"][1], "to_table": "parks", "to_columns": ["id"], "columns": ["gate_id"]}
|
||||
out = catalog.apply_human_edit(existing, {"relations": [moved]}, table="visits", at=AT)
|
||||
assert out["relations"][0]["id"] == catalog.relation_id("visits", ["gate_id"], "parks", ["id"])
|
||||
assert out["relations"][0]["source"] == "human"
|
||||
@@ -0,0 +1,187 @@
|
||||
"""数据目录交给模型的文本、关系图与连接路径、冻结内容、使用次数。"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
from app.data import catalog
|
||||
from app.data.catalog import make_item
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Artifact, DataSource, Run
|
||||
|
||||
MARK = "（推断，未确认）"
|
||||
|
||||
|
||||
def _rel(table, columns, to_table, to_columns, source="fk", status=None, cardinality="many_to_one"):
|
||||
return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns, "to_table": to_table,
|
||||
"to_columns": to_columns, "cardinality": cardinality, "coverage": None, "source": source,
|
||||
"status": status or catalog.initial_status(source)}
|
||||
|
||||
|
||||
VISITS = {
|
||||
"label": make_item("入园记录", "human"),
|
||||
"grain": make_item("每张门票每次检票一行", "llm"),
|
||||
"kind": make_item("fact", "llm", "confirmed"),
|
||||
"keys": make_item(["ticket_no"], "human"),
|
||||
"business_date": make_item({"column": "visit_time", "rule": "按检票时间计", "timezone": "+08:00"}, "human"),
|
||||
"valid_filter": make_item("status = 1", "human"),
|
||||
"dedup": make_item("同一票号只计一次", "llm", "rejected"),
|
||||
"columns": {
|
||||
"visitor_count": {"meaning": make_item("这次检票进园的人数", "human"), "unit": make_item("人", "llm"),
|
||||
"measure": make_item("flow", "llm", "confirmed")},
|
||||
"status": {"codes": make_item({"1": "有效", "0": "作废"}, "human")},
|
||||
},
|
||||
"relations": [_rel("visits", ["park_id"], "parks", ["id"]),
|
||||
_rel("visits", ["gate_id"], "gates", ["id"], source="name"),
|
||||
_rel("visits", ["member_id"], "members", ["id"], source="name", status="rejected")],
|
||||
}
|
||||
|
||||
|
||||
def test_full_render_shows_every_fact_and_marks_inferences():
|
||||
text = catalog.render_table_notes(VISITS)
|
||||
assert "中文名：入园记录" in text
|
||||
assert f"粒度：每张门票每次检票一行{MARK}" in text
|
||||
assert "表类型：事实表" in text and "业务主键：ticket_no" in text
|
||||
assert "业务日期：visit_time" in text and "按检票时间计" in text and "+08:00" in text
|
||||
assert "有效记录条件：status = 1" in text
|
||||
assert "visitor_count：这次检票进园的人数" in text and f"单位：人{MARK}" in text and "流量" in text
|
||||
assert "1=有效" in text and "0=作废" in text
|
||||
assert "park_id → parks.id" in text and f"gate_id → gates.id（多对一）{MARK}" in text
|
||||
# 驳回的项模型看不到
|
||||
assert "同一票号只计一次" not in text and "members" not in text
|
||||
# 有确证、已确认的不标推断
|
||||
assert "表类型：事实表（" not in text and "中文名：入园记录（" not in text
|
||||
|
||||
|
||||
def test_render_of_empty_or_all_rejected_notes_is_empty():
|
||||
assert catalog.render_table_notes({}) == ""
|
||||
assert catalog.render_table_notes({"dedup": make_item("x", "llm", "rejected")}) == ""
|
||||
|
||||
|
||||
def test_render_beside_system_notes_only_adds_what_they_do_not_cover():
|
||||
"""导入表格：表和列已有按核对结果生成的说明。保留说明，只追加人工确认过的、或说明没有覆盖的项。"""
|
||||
meta = {"comment": "粒度：每个日期。客流 等于 各入口之和（导入时已逐日核对）",
|
||||
"columns": [{"name": "日期", "type": "TEXT", "comment": "格式 YYYY-MM-DD"},
|
||||
{"name": "客流", "type": "INTEGER", "comment": None}]}
|
||||
notes = {"grain": make_item("每天一行", "llm"), "label": make_item("客流日报", "llm"),
|
||||
"description": make_item("人工补充的说明", "human"),
|
||||
"columns": {"日期": {"meaning": make_item("统计日期", "llm"), "measure": make_item("attribute", "llm")},
|
||||
"客流": {"unit": make_item("人次", "llm")}}}
|
||||
text = catalog.render_table_notes(notes, meta=meta, system_notes=True)
|
||||
assert "每天一行" not in text and "统计日期" not in text # 说明已经覆盖、又没人确认的：不追加
|
||||
assert "客流日报" in text and "人工补充的说明" in text # 说明没有覆盖的、人工确认过的：追加
|
||||
assert "属性" in text and "人次" in text
|
||||
|
||||
|
||||
def test_compact_index_one_line_per_table():
|
||||
cache = {"tables": {"visits": {"qualified": "visits", "columns": []},
|
||||
"parks": {"qualified": "main.parks", "columns": []},
|
||||
"gates": {"qualified": "gates", "columns": []}}}
|
||||
text = catalog.render_table_index(cache, {"visits": VISITS, "parks": {"label": make_item("景区", "llm")}})
|
||||
lines = text.splitlines()
|
||||
assert len(lines) == 3
|
||||
assert lines[0].startswith("visits｜入园记录｜每张门票每次检票一行") and lines[0].endswith(MARK)
|
||||
assert lines[1].startswith("main.parks｜景区") and MARK in lines[1]
|
||||
assert lines[2] == "gates"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 关系图
|
||||
|
||||
|
||||
NOTES = {
|
||||
"visits": {"relations": [_rel("visits", ["park_id"], "parks", ["id"]),
|
||||
_rel("visits", ["member_id"], "members", ["id"], source="name", status="rejected")]},
|
||||
"channel_visits": {"relations": [_rel("channel_visits", ["channel_id"], "channels", ["id"]),
|
||||
_rel("channel_visits", ["visit_id"], "visits", ["id"])]},
|
||||
"orders": {"relations": [_rel("orders", ["channel_id"], "channels", ["id"]),
|
||||
_rel("orders", ["member_id"], "members", ["id"])]},
|
||||
}
|
||||
|
||||
|
||||
def test_graph_adjacency_is_bidirectional_and_skips_rejected():
|
||||
graph = catalog.relation_graph(NOTES)
|
||||
assert {e.to_table for e in graph["visits"]} == {"parks", "channel_visits"}
|
||||
back = next(e for e in graph["channels"] if e.to_table == "orders")
|
||||
assert back.from_columns == ("id",) and back.to_columns == ("channel_id",) and back.cardinality == "one_to_many"
|
||||
assert all(e.to_table != "members" for e in graph["visits"])
|
||||
|
||||
|
||||
def test_join_paths_up_to_two_hops():
|
||||
graph = catalog.relation_graph(NOTES)
|
||||
paths = catalog.join_paths(graph, "visits", "channels")
|
||||
assert [tuple(e.to_table for e in p) for p in paths] == [("channel_visits", "channels")]
|
||||
assert paths[0][0].condition() == "visits.id = channel_visits.visit_id"
|
||||
assert paths[0][1].condition() == "channel_visits.channel_id = channels.id"
|
||||
assert catalog.join_paths(graph, "visits", "parks")[0][0].to_table == "parks"
|
||||
# 三跳以上不给：visits → channel_visits → channels → orders
|
||||
assert catalog.join_paths(graph, "visits", "orders") == []
|
||||
# 驳回的关系不出现在路径里
|
||||
assert catalog.join_paths(graph, "visits", "members") == []
|
||||
|
||||
|
||||
def test_graph_can_fill_in_relations_from_structure():
|
||||
"""没起草过目录的表，关系图可以按表结构现推（外键 + 命名推断），驳回过的照样排除。"""
|
||||
cache = {"tables": {
|
||||
"visits": {"columns": [{"name": "id", "type": "INTEGER"}, {"name": "gate_id", "type": "INTEGER"},
|
||||
{"name": "member_id", "type": "INTEGER"}],
|
||||
"primary_key": ["id"], "foreign_keys": []},
|
||||
"gates": {"columns": [{"name": "id", "type": "INTEGER"}], "primary_key": ["id"], "foreign_keys": []},
|
||||
"members": {"columns": [{"name": "id", "type": "INTEGER"}], "primary_key": ["id"], "foreign_keys": []},
|
||||
}}
|
||||
rejected = {"visits": {"relations": [_rel("visits", ["member_id"], "members", ["id"], source="name",
|
||||
status="rejected")]}}
|
||||
graph = catalog.relation_graph(rejected, schema_cache=cache)
|
||||
assert {e.to_table for e in graph["visits"]} == {"gates"}
|
||||
assert graph["gates"][0].status == "proposed"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 冻结
|
||||
|
||||
|
||||
def test_frozen_catalog_strips_rejected_and_keeps_versions():
|
||||
entries = {"visits": catalog.CatalogEntry("visits", VISITS, 4),
|
||||
"gone": catalog.CatalogEntry("gone", {"label": make_item("x", "human")}, 2),
|
||||
"empty": catalog.CatalogEntry("empty", {"dedup": make_item("x", "llm", "rejected")}, 1)}
|
||||
frozen = catalog.frozen_catalog(entries, ["visits", "empty", "parks"])
|
||||
assert set(frozen) == {"visits"} # 表结构里没有的、全被驳回的都不冻结
|
||||
assert frozen["visits"]["version"] == 4
|
||||
assert "dedup" not in frozen["visits"]["notes"]
|
||||
assert len(frozen["visits"]["notes"]["relations"]) == 2
|
||||
assert catalog.frozen_catalog({}, ["visits"]) == {}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 使用次数
|
||||
|
||||
|
||||
async def test_table_usage_counts_run_queries_per_table(monkeypatch):
|
||||
sid = uuid.uuid4().hex
|
||||
name = f"use_{sid[:8]}"
|
||||
cache = {"tables": {"visits": {"qualified": "main.visits"}, "parks": {"qualified": "main.parks"},
|
||||
"明细": {"qualified": "main.明细"}}}
|
||||
run_a, run_b = uuid.uuid4().hex, uuid.uuid4().hex
|
||||
legacy = {sid[:8] + "legacy1": {"source": name, "sql": 'SELECT * FROM "Visits" v JOIN main.parks p ON 1=1'}}
|
||||
monkeypatch.setattr(catalog, "_load_artifact", lambda aid: legacy.get(aid))
|
||||
async with SessionLocal() as session:
|
||||
session.add(DataSource(id=sid, name=name, kind="sqlite", database=":memory:", schema_cache=cache))
|
||||
session.add_all([Run(id=run_a, graph={}), Run(id=run_b, graph={})])
|
||||
session.add_all([
|
||||
# 新写法：查询当时把源和表名记在工件引用的 meta 里
|
||||
Artifact(id=sid[:8] + "q1", run_id=run_a, kind="query_snapshot",
|
||||
meta={"source_id": sid, "source": name, "tables": ["visits"]}),
|
||||
Artifact(id=sid[:8] + "q2", run_id=run_b, kind="query_snapshot",
|
||||
meta={"source_id": sid, "source": name, "tables": ["visits", "明细"]}),
|
||||
# 同一个查询快照被两次运行引用：各算一次
|
||||
Artifact(id=sid[:8] + "q2", run_id=run_a, kind="query_snapshot",
|
||||
meta={"source_id": sid, "source": name, "tables": ["visits", "明细"]}),
|
||||
# 别的数据源、工具库试用（不属于任何运行）、别的工件：不算
|
||||
Artifact(id=sid[:8] + "q3", run_id=run_a, kind="query_snapshot",
|
||||
meta={"source_id": "other", "source": "other", "tables": ["visits"]}),
|
||||
Artifact(id=sid[:8] + "q4", run_id="playground", kind="query_snapshot",
|
||||
meta={"source_id": sid, "source": name, "tables": ["visits"]}),
|
||||
Artifact(id=sid[:8] + "s1", run_id=run_a, kind="schema_snapshot", meta={"source": name}),
|
||||
# 老的查询快照引用没有 meta：读工件内容里的源名和 SQL
|
||||
Artifact(id=sid[:8] + "legacy1", run_id=run_b, kind="query_snapshot", meta={}),
|
||||
])
|
||||
await session.commit()
|
||||
source = await session.get(DataSource, sid)
|
||||
usage = await catalog.table_usage(session, source)
|
||||
assert usage == {"visits": 4, "明细": 2, "parks": 1}
|
||||
@@ -0,0 +1,185 @@
|
||||
"""数据目录在运行时：db_schema 工具输出附上目录，查询时目录随表结构快照冻结进证据，查询快照记下用到的表。
|
||||
|
||||
- describe_table（db_schema 工具的输出）附上这张表的目录；推断的项标「推断，未确认」，驳回的不出现。
|
||||
导入表格的源保留系统生成的说明，目录只追加说明没有覆盖的、或人工确认过的项。
|
||||
- 表结构快照新增顶层键 catalog：{表: {version, notes（去掉驳回项）}}。没有目录的源一个字都不加；
|
||||
目录不变则快照不变（内容寻址，哈希相同）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
|
||||
from app.core.artifact_store import canonical_json, content_hash, load
|
||||
from app.data import catalog
|
||||
from app.data.catalog import make_item
|
||||
from app.data.engine import engines
|
||||
from app.data.introspect import describe_table, introspect
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import Artifact, DataSource
|
||||
from app.tools.datasource import _store_schema, build_datasource_tools
|
||||
from app.tools.registry import ToolContext
|
||||
|
||||
MARK = "（推断，未确认）"
|
||||
|
||||
|
||||
def _rel(table, columns, to_table, to_columns, source="fk", status=None):
|
||||
return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns, "to_table": to_table,
|
||||
"to_columns": to_columns, "cardinality": "many_to_one", "coverage": None, "source": source,
|
||||
"status": status or catalog.initial_status(source)}
|
||||
|
||||
|
||||
CACHE = {
|
||||
"schema": "main", "synced_at": "2026-10-01T00:00:00+00:00", "truncated": False, "total": 2,
|
||||
"tables": {
|
||||
"visits": {"qualified": "visits", "schema": None, "is_view": False, "comment": None, "primary_key": ["id"],
|
||||
"columns": [{"name": "id", "type": "INTEGER", "nullable": False, "comment": None},
|
||||
{"name": "park_id", "type": "INTEGER", "nullable": False, "comment": None},
|
||||
{"name": "visitor_count", "type": "INTEGER", "nullable": False, "comment": None},
|
||||
{"name": "status", "type": "INTEGER", "nullable": False, "comment": None}]},
|
||||
"parks": {"qualified": "parks", "schema": None, "is_view": False, "comment": None, "primary_key": ["id"],
|
||||
"columns": [{"name": "id", "type": "INTEGER", "nullable": False, "comment": None}]},
|
||||
},
|
||||
}
|
||||
|
||||
NOTES = {
|
||||
"label": make_item("入园记录", "human"),
|
||||
"grain": make_item("每张门票每次检票一行", "llm"),
|
||||
"keys": make_item(["id"], "human"),
|
||||
"business_date": make_item({"column": "visit_time", "rule": "按检票时间计"}, "human"),
|
||||
"valid_filter": make_item("status = 1", "llm"),
|
||||
"dedup": make_item("按票号去重", "llm", "rejected"),
|
||||
"columns": {"visitor_count": {"meaning": make_item("这次检票进园的人数", "llm"), "unit": make_item("人", "human"),
|
||||
"measure": make_item("flow", "llm")},
|
||||
"status": {"codes": make_item({"1": "有效", "0": "作废"}, "human")}},
|
||||
"relations": [_rel("visits", ["park_id"], "parks", ["id"])],
|
||||
}
|
||||
|
||||
|
||||
def _source(cache=CACHE, **kw) -> SimpleNamespace:
|
||||
return SimpleNamespace(id=kw.pop("id", "src-x"), name=kw.pop("name", "scenic"), kind="sqlite", options={},
|
||||
readonly=True, description="", schema_cache=cache, origin=kw.pop("origin", "manual"), **kw)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- describe_table
|
||||
|
||||
|
||||
def test_describe_table_appends_catalog():
|
||||
text = describe_table(_source(), "visits", catalog={"visits": NOTES})
|
||||
head, _, block = text.partition(catalog.NOTES_HEADING)
|
||||
assert head.startswith("表 visits") and " visitor_count INTEGER" in head # 原来的列清单照旧
|
||||
assert "中文名：入园记录" in block and f"粒度：每张门票每次检票一行{MARK}" in block
|
||||
assert "业务主键：id" in block and "业务日期：visit_time；规则：按检票时间计" in block
|
||||
assert f"有效记录条件：status = 1{MARK}" in block
|
||||
assert f"visitor_count：这次检票进园的人数{MARK}；单位：人；度量类型：流量，可跨期加总{MARK}" in block
|
||||
assert "status：码值：1=有效、0=作废" in block
|
||||
assert "park_id → parks.id（多对一）" in block
|
||||
assert "按票号去重" not in text
|
||||
|
||||
|
||||
def test_describe_table_without_catalog_is_unchanged():
|
||||
plain = describe_table(_source(), "visits")
|
||||
assert describe_table(_source(), "visits", catalog={}) == plain
|
||||
assert describe_table(_source(), "visits", catalog={"parks": NOTES}) == plain
|
||||
assert catalog.NOTES_HEADING not in plain
|
||||
# 全名、大小写不一致也对得上目录
|
||||
assert catalog.NOTES_HEADING in describe_table(_source(), "MAIN.Visits", catalog={"visits": NOTES})
|
||||
|
||||
|
||||
def test_describe_table_keeps_system_notes_of_imported_tables():
|
||||
"""导入表格：说明是按核对结果生成的。保留说明，目录不重复它、不和它抢话。"""
|
||||
cache = {"import_mode": "recipe", "tables": {"日报": {
|
||||
"qualified": "日报", "comment": "粒度：每个日期。不要把各入口相加", "primary_key": [],
|
||||
"columns": [{"name": "日期", "type": "TEXT", "nullable": False, "comment": "格式 YYYY-MM-DD"},
|
||||
{"name": "客流", "type": "INTEGER", "nullable": True, "comment": "单位：人次"}]}}}
|
||||
notes = {"label": make_item("客流日报", "llm"), "grain": make_item("每天一行", "llm"),
|
||||
"columns": {"客流": {"unit": make_item("人", "llm"), "measure": make_item("flow", "llm")},
|
||||
"日期": {"meaning": make_item("统计日期", "human")}}}
|
||||
text = describe_table(_source(cache, origin="upload"), "日报", catalog={"日报": notes})
|
||||
assert text.startswith("表 日报（粒度：每个日期。不要把各入口相加）")
|
||||
assert "单位：人次" in text and "格式 YYYY-MM-DD" in text # 系统说明原样保留
|
||||
assert "每天一行" not in text # 表说明已写粒度：推断的粒度不追加
|
||||
assert "单位：人" + MARK not in text # 列说明已写单位：推断的单位不追加
|
||||
assert "客流日报" in text and "流量" in text # 说明没有覆盖的照常追加
|
||||
assert "日期：统计日期" in text # 人工确认过的追加
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 冻结进表结构快照
|
||||
|
||||
|
||||
def _entry(notes, version=1):
|
||||
return {"visits": catalog.CatalogEntry("visits", notes, version)}
|
||||
|
||||
|
||||
async def test_store_schema_freezes_catalog_and_is_content_addressed():
|
||||
ctx = SimpleNamespace(run_id="", node_id="")
|
||||
plain = await _store_schema(_source(), ctx)
|
||||
before = {"source": "scenic", "tables": CACHE["tables"],
|
||||
**{k: CACHE[k] for k in ("schema", "synced_at", "truncated", "total")}}
|
||||
assert plain == content_hash(canonical_json(before)) # 没有目录：和以前一字不差
|
||||
assert await _store_schema(_source(), ctx, catalog={}) == plain
|
||||
|
||||
first = await _store_schema(_source(), ctx, catalog=_entry(NOTES, 3))
|
||||
frozen = load(first)
|
||||
assert frozen["catalog"]["visits"]["version"] == 3
|
||||
assert "dedup" not in frozen["catalog"]["visits"]["notes"] # 驳回的不冻结
|
||||
assert frozen["tables"] == CACHE["tables"] # 已有字段的语义不变
|
||||
# 目录不变：同一件工件
|
||||
assert await _store_schema(_source(), ctx, catalog=_entry(json.loads(json.dumps(NOTES)), 3)) == first
|
||||
# 目录变了（升了版本）：换一件
|
||||
assert await _store_schema(_source(), ctx, catalog=_entry(NOTES, 4)) != first
|
||||
# 只有驳回项、或者表不在快照里：不加这个键
|
||||
assert await _store_schema(_source(), ctx, catalog={
|
||||
"visits": catalog.CatalogEntry("visits", {"dedup": make_item("x", "llm", "rejected")}, 1),
|
||||
"gone": catalog.CatalogEntry("gone", {"label": make_item("x", "human")}, 1)}) == plain
|
||||
|
||||
|
||||
def test_snapshot_readers_ignore_catalog_key():
|
||||
"""读表结构快照的地方（证据台账的表结构条目、裁判摘录的口径行）只认自己的键，多一个 catalog 不影响。"""
|
||||
from app.data.provenance import judge_lines
|
||||
from app.engine.evidence import schema_entry_fields
|
||||
|
||||
tables = {"日报": {"qualified": "日报", "comment": "粒度：每个日期", "primary_key": [],
|
||||
"columns": [{"name": "客流", "type": "INTEGER", "nullable": True, "comment": "单位：人次"}]}}
|
||||
base = {"source": "flow", "tables": tables, "schema": "main"}
|
||||
with_catalog = {**base, "catalog": {"日报": {"version": 2, "notes": {"label": make_item("客流日报", "human")}}}}
|
||||
assert schema_entry_fields(with_catalog) == schema_entry_fields(base)
|
||||
query = {"data_version": "snap1", "columns": ["客流"], "sql": "SELECT 客流 FROM 日报"}
|
||||
lines = judge_lines(query, base, lambda _id: None, tables=["日报"], hidden=set())
|
||||
assert lines and judge_lines(query, with_catalog, lambda _id: None, tables=["日报"], hidden=set()) == lines
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 工具里串起来
|
||||
|
||||
|
||||
async def test_tools_show_and_freeze_catalog_and_record_query_tables(scenic_db):
|
||||
async with SessionLocal() as session:
|
||||
row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=scenic_db)
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
row.schema_cache = await introspect(row)
|
||||
await session.commit()
|
||||
await catalog.write_entry(session, row.id, "visits", {"label": make_item("入园记录", "human"),
|
||||
"grain": make_item("每张门票每次检票一行", "llm")},
|
||||
if_version=0, actor="王敏")
|
||||
run_id = uuid.uuid4().hex
|
||||
ctx = ToolContext(run_id=run_id, node_id="n1", data_versions=None)
|
||||
tools = {t.name: t for t in await build_datasource_tools(
|
||||
[f"db_query__{row.name}", f"db_schema__{row.name}"], ctx, session)}
|
||||
try:
|
||||
shown = await tools[f"db_schema__{row.name}"].coroutine(table="visits")
|
||||
assert "中文名：入园记录" in shown and f"粒度：每张门票每次检票一行{MARK}" in shown
|
||||
other = await tools[f"db_schema__{row.name}"].coroutine(table="parks")
|
||||
assert catalog.NOTES_HEADING not in other
|
||||
|
||||
result = json.loads(await tools[f"db_query__{row.name}"].coroutine(
|
||||
sql="SELECT v.id FROM visits v JOIN parks p ON p.id = v.park_id LIMIT 3"))
|
||||
frozen = load(result["schema_artifact"])
|
||||
assert frozen["catalog"] == {"visits": {"version": 1, "notes": {
|
||||
"label": make_item("入园记录", "human"), "grain": make_item("每张门票每次检票一行", "llm")}}}
|
||||
async with SessionLocal() as session:
|
||||
ref = await session.get(Artifact, (result["artifact"], run_id))
|
||||
assert ref.meta == {"source_id": row.id, "source": row.name, "tables": ["visits", "parks"]}
|
||||
finally:
|
||||
await engines.invalidate(row.id)
|
||||
@@ -0,0 +1,147 @@
|
||||
"""数据目录的存储：catalog_notes 表、版本号与乐观锁、结构校验。
|
||||
|
||||
目录是多人可改的长期资产（助手起草、人工确认、阶段 2 的数据剖析都往里写），所以每次写入必须带上
|
||||
「我是基于第几版改的」：两个人同时打开同一张表各改各的，后提交的那个要被拦下来重新载入，不能悄悄
|
||||
盖掉前一个人的确认。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from app.data import catalog
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import CatalogNote, DataSource
|
||||
|
||||
|
||||
async def _source() -> str:
|
||||
async with SessionLocal() as session:
|
||||
row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=":memory:")
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
return row.id
|
||||
|
||||
|
||||
def _label(value: str, source: str = "human", status: str = "confirmed") -> dict:
|
||||
return {"value": value, "source": source, "status": status}
|
||||
|
||||
|
||||
async def test_first_write_creates_version_one_and_records_actor():
|
||||
sid = await _source()
|
||||
async with SessionLocal() as session:
|
||||
entry = await catalog.write_entry(session, sid, "visits", {"label": _label("入园记录")},
|
||||
if_version=0, actor="王敏")
|
||||
assert entry.version == 1
|
||||
assert entry.updated_by == "王敏"
|
||||
assert entry.notes["label"]["value"] == "入园记录"
|
||||
async with SessionLocal() as session:
|
||||
again = await catalog.read_entry(session, sid, "visits")
|
||||
assert again is not None and again.version == 1 and again.updated_at is not None
|
||||
|
||||
|
||||
async def test_each_write_bumps_version_and_stale_version_conflicts():
|
||||
sid = await _source()
|
||||
async with SessionLocal() as session:
|
||||
await catalog.write_entry(session, sid, "visits", {"label": _label("入园记录")}, if_version=0, actor=None)
|
||||
second = await catalog.write_entry(session, sid, "visits", {"label": _label("入园明细")},
|
||||
if_version=1, actor="李雷")
|
||||
assert second.version == 2
|
||||
# 另一个人还拿着第 1 版：被拦下，内容不变
|
||||
with pytest.raises(catalog.CatalogConflict):
|
||||
await catalog.write_entry(session, sid, "visits", {"label": _label("别的名字")}, if_version=1, actor="韩梅")
|
||||
# 以为这张表还没有目录（if_version=0），其实已经有了：同样是冲突
|
||||
with pytest.raises(catalog.CatalogConflict):
|
||||
await catalog.write_entry(session, sid, "visits", {"label": _label("别的名字")}, if_version=0, actor=None)
|
||||
async with SessionLocal() as session:
|
||||
entry = await catalog.read_entry(session, sid, "visits")
|
||||
assert entry.version == 2 and entry.notes["label"]["value"] == "入园明细" and entry.updated_by == "李雷"
|
||||
|
||||
|
||||
async def test_unchanged_notes_do_not_bump_version():
|
||||
"""内容没变就不升版本：冻结进表结构快照的是 {版本, 内容}，空写一次也升版本的话，快照哈希会无故变化。"""
|
||||
sid = await _source()
|
||||
notes = {"label": _label("渠道")}
|
||||
async with SessionLocal() as session:
|
||||
first = await catalog.write_entry(session, sid, "channels", notes, if_version=0, actor="甲")
|
||||
same = await catalog.write_entry(session, sid, "channels", dict(notes), if_version=1, actor="乙")
|
||||
assert same.version == first.version == 1
|
||||
assert same.updated_by == "甲"
|
||||
|
||||
|
||||
async def test_unique_per_source_and_table():
|
||||
sid = await _source()
|
||||
async with SessionLocal() as session:
|
||||
session.add(CatalogNote(source_id=sid, table_name="visits", notes={}, version=1))
|
||||
await session.commit()
|
||||
session.add(CatalogNote(source_id=sid, table_name="visits", notes={}, version=1))
|
||||
with pytest.raises(IntegrityError):
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def test_deleting_source_cascades():
|
||||
sid = await _source()
|
||||
async with SessionLocal() as session:
|
||||
await catalog.write_entry(session, sid, "visits", {"label": _label("入园记录")}, if_version=0, actor=None)
|
||||
row = await session.get(DataSource, sid)
|
||||
await session.delete(row)
|
||||
await session.commit()
|
||||
async with SessionLocal() as session:
|
||||
left = (await session.execute(select(CatalogNote).where(CatalogNote.source_id == sid))).scalars().all()
|
||||
assert left == []
|
||||
|
||||
|
||||
async def test_read_catalog_lists_all_tables_of_a_source():
|
||||
sid, other = await _source(), await _source()
|
||||
async with SessionLocal() as session:
|
||||
await catalog.write_entry(session, sid, "visits", {"label": _label("入园记录")}, if_version=0, actor=None)
|
||||
await catalog.write_entry(session, sid, "stores", {"label": _label("门店")}, if_version=0, actor=None)
|
||||
await catalog.write_entry(session, other, "visits", {"label": _label("别处")}, if_version=0, actor=None)
|
||||
entries = await catalog.read_catalog(session, sid)
|
||||
assert set(entries) == {"visits", "stores"}
|
||||
assert entries["stores"].notes["label"]["value"] == "门店"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("notes, fragment", [
|
||||
({"label": "入园记录"}, "中文名"), # 不是项
|
||||
({"label": {"value": "x", "source": "guess", "status": "proposed"}}, "来源"), # 来源不在取值里
|
||||
({"label": {"value": "x", "source": "llm", "status": "maybe"}}, "状态"),
|
||||
({"kind": _label("table")}, "表类型"),
|
||||
({"keys": _label("visit_id")}, "业务主键"), # 应为字符串列表
|
||||
({"business_date": _label({"column": "visit_date", "rule": 1})}, "业务日期"),
|
||||
({"columns": {"amount": {"measure": _label("sum")}}}, "列 amount 的度量类型"),
|
||||
({"columns": {"status": {"codes": _label({"1": 2})}}}, "列 status 的码值"),
|
||||
({"formula": _label("sum(amount)")}, "formula"), # 目录不放公式
|
||||
({"relations": [{"id": "r1", "columns": ["a"], "to_table": "t", "to_columns": [],
|
||||
"source": "fk", "status": "verified"}]}, "关联关系 r1"),
|
||||
])
|
||||
def test_validate_rejects_malformed_notes(notes, fragment):
|
||||
problems = catalog.validate_notes(notes)
|
||||
assert problems, notes
|
||||
assert any(fragment in p for p in problems), problems
|
||||
|
||||
|
||||
def test_validate_accepts_full_example():
|
||||
notes = {
|
||||
"label": _label("入园记录"), "description": _label("每次检票入园记一行", "llm", "proposed"),
|
||||
"grain": _label("每次入园"), "keys": _label(["visit_id"]), "kind": _label("fact"),
|
||||
"business_date": _label({"column": "visit_date", "rule": "按检票时间计", "timezone": "+08:00"}),
|
||||
"valid_filter": _label("status = 1"), "dedup": _label("同一票号只计一次"),
|
||||
"columns": {"amount": {"label": _label("实收金额"), "meaning": _label("扣除优惠后的金额"),
|
||||
"unit": _label("元"), "measure": _label("flow"),
|
||||
"codes": _label({"1": "有效"}, "comment", "proposed")}},
|
||||
"relations": [{"id": catalog.relation_id("visits", ["store_id"], "stores", ["id"]), "columns": ["store_id"],
|
||||
"to_table": "stores", "to_columns": ["id"], "cardinality": "many_to_one", "coverage": None,
|
||||
"source": "fk", "status": "verified"}],
|
||||
}
|
||||
assert catalog.validate_notes(notes) == []
|
||||
|
||||
|
||||
async def test_write_rejects_invalid_notes():
|
||||
sid = await _source()
|
||||
async with SessionLocal() as session:
|
||||
with pytest.raises(catalog.CatalogInvalid) as e:
|
||||
await catalog.write_entry(session, sid, "visits", {"label": "入园记录"}, if_version=0, actor=None)
|
||||
assert e.value.problems
|
||||
@@ -0,0 +1,119 @@
|
||||
"""探查时补读外键约束和唯一约束 / 唯一索引，以及数据目录在合成大库上的起草效果。
|
||||
|
||||
数据目录的关系起草靠它：外键记成有确证的关系，唯一约束用来判断一对一。读不到（方言不支持、权限不够、
|
||||
某个对象反射报错）就跳过那一项，不能毁掉整次探查；跳过的表不写这两个键，下游据此分得清「读过、没有」
|
||||
和「不知道」（旧缓存里也没有这两个键）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.engine.reflection import Inspector
|
||||
|
||||
from app.data import catalog
|
||||
from app.data.engine import engines
|
||||
from app.data.introspect import introspect
|
||||
from tests.fixtures.catalog import scenic
|
||||
|
||||
|
||||
def _source(path: str, sid: str = "scenic-src") -> SimpleNamespace:
|
||||
return SimpleNamespace(id=sid, name="scenic", kind="sqlite", database=path, host=None, port=None, username=None,
|
||||
password=None, options={}, readonly=True, description="景区业务库", schema_cache={},
|
||||
origin="manual")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def scenic_cache(scenic_db):
|
||||
source = _source(scenic_db, "scenic-cache")
|
||||
try:
|
||||
cache = await introspect(source)
|
||||
finally:
|
||||
await engines.invalidate(source.id)
|
||||
return cache
|
||||
|
||||
|
||||
async def test_fixture_is_large_and_complete(scenic_cache):
|
||||
tables = scenic_cache["tables"]
|
||||
assert len(tables) == scenic.TABLE_COUNT + len(scenic.VIEWS) > 40
|
||||
assert not scenic_cache.get("truncated") and not scenic_cache.get("skipped")
|
||||
|
||||
|
||||
async def test_foreign_keys_are_read(scenic_cache):
|
||||
found = {(name, tuple(fk["columns"]), fk["to_table"], tuple(fk["to_columns"]))
|
||||
for name, meta in scenic_cache["tables"].items() for fk in meta["foreign_keys"]}
|
||||
assert found == scenic.FK_RELATIONS
|
||||
# 每张表、每个视图都读过：没有外键的是空列表，不是缺键
|
||||
assert all(isinstance(m.get("foreign_keys"), list) for m in scenic_cache["tables"].values())
|
||||
assert scenic_cache["tables"]["v_daily_visits"]["foreign_keys"] == []
|
||||
|
||||
|
||||
async def test_unique_constraints_and_unique_indexes_are_read(scenic_cache):
|
||||
tables = scenic_cache["tables"]
|
||||
for name, groups in scenic.UNIQUE.items():
|
||||
for group in groups:
|
||||
assert group in tables[name]["unique"], (name, group)
|
||||
# 主键不重复记成唯一约束；没有唯一约束的是空列表
|
||||
assert ["id"] not in tables["orders"]["unique"]
|
||||
assert tables["refunds"]["unique"] == []
|
||||
|
||||
|
||||
async def test_constraint_reflection_failure_does_not_break_introspection(scenic_db, monkeypatch):
|
||||
def boom(*args, **kwargs):
|
||||
raise RuntimeError("permission denied for information_schema")
|
||||
|
||||
for name in ("get_multi_foreign_keys", "get_foreign_keys", "get_multi_unique_constraints",
|
||||
"get_unique_constraints", "get_multi_indexes", "get_indexes"):
|
||||
monkeypatch.setattr(Inspector, name, boom)
|
||||
source = _source(scenic_db, "scenic-broken")
|
||||
try:
|
||||
cache = await introspect(source)
|
||||
finally:
|
||||
await engines.invalidate(source.id)
|
||||
assert not cache.get("failed") and len(cache["tables"]) == scenic.TABLE_COUNT + len(scenic.VIEWS)
|
||||
visits = cache["tables"]["visits"]
|
||||
assert "foreign_keys" not in visits and "unique" not in visits
|
||||
assert visits["primary_key"] == ["id"] and visits["columns"]
|
||||
|
||||
|
||||
async def test_not_implemented_constraints_are_skipped(scenic_db, monkeypatch):
|
||||
def unsupported(*args, **kwargs):
|
||||
raise NotImplementedError
|
||||
|
||||
monkeypatch.setattr(Inspector, "get_multi_foreign_keys", unsupported)
|
||||
monkeypatch.setattr(Inspector, "get_foreign_keys", unsupported)
|
||||
source = _source(scenic_db, "scenic-nofk")
|
||||
try:
|
||||
cache = await introspect(source)
|
||||
finally:
|
||||
await engines.invalidate(source.id)
|
||||
assert "foreign_keys" not in cache["tables"]["visits"]
|
||||
assert isinstance(cache["tables"]["visits"]["unique"], list)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 在合成大库上起草
|
||||
|
||||
|
||||
async def test_structure_draft_on_scenic_db(scenic_db, scenic_cache):
|
||||
source = SimpleNamespace(**{**vars(_source(scenic_db)), "schema_cache": scenic_cache})
|
||||
fk, name = set(), set()
|
||||
for table in scenic_cache["tables"]:
|
||||
for rel in catalog.draft_structure(source, table).notes.get("relations", []):
|
||||
entry = (table, tuple(rel["columns"]), rel["to_table"], tuple(rel["to_columns"]))
|
||||
(fk if rel["source"] == "fk" else name).add(entry)
|
||||
assert rel["status"] == ("verified" if rel["source"] == "fk" else "proposed")
|
||||
assert fk == scenic.FK_RELATIONS
|
||||
assert name == scenic.NAME_RELATIONS
|
||||
inferred_columns = {(t, c[0]) for t, c, _, _ in name}
|
||||
assert not inferred_columns & set(scenic.NOT_INFERRED)
|
||||
|
||||
|
||||
async def test_join_path_through_link_table(scenic_cache):
|
||||
graph = catalog.relation_graph({}, schema_cache=scenic_cache)
|
||||
paths = catalog.join_paths(graph, "visits", "channels")
|
||||
via = [tuple(e.to_table for e in p) for p in paths]
|
||||
assert ("channel_visits", "channels") in via
|
||||
best = paths[0]
|
||||
assert all(e.status == "verified" for e in best) # 两跳都是外键，排在推断的路径前面
|
||||
assert [e.condition() for e in best] == ["visits.id = channel_visits.visit_id",
|
||||
"channel_visits.channel_id = channels.id"]
|
||||
Reference in New Issue
Block a user