mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
schema 发现改成看事实，不再靠黑名单猜
起因是一个合理的质疑：这套 schema 处理是不是只对那一个 Oracle 有效。查证下来 "schema" 在四家里根本不是同一个东西——MySQL 的 schema 就是 database （SHOW schemas），Oracle 的 schema 就是 user（all_users），PG 是库下面的 命名空间（pg_namespace），SQLite 只有 main。SQLAlchemy 抹平了调用方式， 没抹平语义。 ## 黑名单方案实测不可用 原来靠一张硬编码的系统 schema 名单过滤。在真实 Oracle 上：94 个用户过滤完还剩 75 个（anonymous、C##OGGADMIN、infa96_domain、xs$null…），人在里面根本找不到 那两个业务 schema。名单再怎么补也补不完——这是在猜，而不是在看。 改成看事实：统计每个 schema 下有多少表/视图，有东西的才留下并按数量排序。 两条路径覆盖全部支持的库——information_schema 是 SQL 标准（MySQL/PG/SQL Server 都有），Oracle 不遵循这条标准，走它自己的 all_tables/all_views。不用"逐个 schema 调 get_table_names"的通用写法：94 个 schema 就是 94 次往返。 黑名单保留但降级为辅助：系统 schema 的对象数往往很大（SYS 有 1871 个）， 光靠排序压不下去。 实测：94 个用户 → 2.3 秒收敛到 ANALYTICS、STAGING 两个业务 schema。 ## schema 填错时给可选项，而不是抛驱动异常 大小写、拼写、或者根本不知道该填什么，都是高频事故。原来直接把 OperationalError 抛出去，等于让人自己猜。现在先看连接本身好不好：能列出候选 就说明只是 schema 不对，这种情况返回空结构 + 候选列表，摘要里直接写 "该账号名下没有对象，数据可能在这些 schema：ANALYTICS、STAGING"。 ## 覆盖 SQLite 没有 information_schema，统计路径安静降级（实测）；没有 schema 概念时 qualified 不硬拼前缀，有 schema 时必须带（Oracle 上少了就是 ORA-00942）。 新增一条用例守着"加新数据库类型时别忘了补统计路径"，漏了就会退化回黑名单猜测。 MySQL / PostgreSQL 手上没有实例，走的是 information_schema 标准路径， 代码路径与 Oracle 共用，但未实测——这点在 README 里会标明。
This commit is contained in:
ThearchyHelios
@@ -39,23 +39,71 @@ _SYSTEM_SCHEMAS = frozenset({
|
||||
})
|
||||
|
||||
|
||||
# 统计每个 schema 下有多少对象。两条路径覆盖全部支持的库：
|
||||
# information_schema 是 SQL 标准（MySQL / PostgreSQL / SQL Server 都有），
|
||||
# Oracle 不遵循这条标准，用它自己的 all_tables / all_views 数据字典。
|
||||
#
|
||||
# 不用"逐个 schema 调 get_table_names"的通用写法：这个库有 94 个 schema，
|
||||
# 那样是 94 次往返，几十秒起步。
|
||||
_SCHEMA_STATS_SQL = {
|
||||
"oracle": """
|
||||
SELECT owner AS s, COUNT(*) AS n FROM (
|
||||
SELECT owner FROM all_tables
|
||||
UNION ALL SELECT owner FROM all_views
|
||||
) GROUP BY owner ORDER BY COUNT(*) DESC
|
||||
""",
|
||||
"_standard": """
|
||||
SELECT table_schema AS s, COUNT(*) AS n
|
||||
FROM information_schema.tables
|
||||
GROUP BY table_schema ORDER BY COUNT(*) DESC
|
||||
""",
|
||||
}
|
||||
|
||||
|
||||
async def schema_stats(source: Any) -> list[tuple[str, int]]:
|
||||
"""每个 schema 有多少表/视图，多的在前。拿不到就返回空表让调用方降级。"""
|
||||
from sqlalchemy import text as _text
|
||||
|
||||
kind = (source.kind or "").lower()
|
||||
if kind == "sqlite":
|
||||
return [] # SQLite 只有 main，没有可选项
|
||||
sql = _SCHEMA_STATS_SQL.get(kind, _SCHEMA_STATS_SQL["_standard"])
|
||||
engine = await engines.get(source)
|
||||
try:
|
||||
async with engine.connect() as conn:
|
||||
rows = (await conn.execute(_text(sql))).fetchall()
|
||||
return [(str(r[0]), int(r[1])) for r in rows if r[0]]
|
||||
except Exception: # noqa: BLE001 - 方言不认就算了，降级到纯名字列表
|
||||
return []
|
||||
|
||||
|
||||
async def list_schemas(source: Any) -> list[str]:
|
||||
"""列出可访问的非系统 schema。
|
||||
"""列出**可能装着业务数据**的 schema，最有可能的排在最前。
|
||||
|
||||
企业环境里只读账号名下往往一张表都没有——数据在别的 schema，靠跨库授权或
|
||||
synonym 访问。直接返回"0 张表"等于把人晾在那儿，得告诉他数据可能在哪。
|
||||
synonym 访问。直接返回"0 张表"等于把人晾在那儿。
|
||||
|
||||
但光列出名字也不够用：实测一个 Oracle 实例有 94 个用户，靠黑名单过滤完
|
||||
还剩 75 个（anonymous、C##OGGADMIN、xs$null…），人在里面找不到那两个
|
||||
业务 schema。所以主要依据不是"猜哪些是系统的"，而是"哪些里面真有东西"——
|
||||
黑名单只用来排掉已知的系统 schema（它们对象数往往还很多，光靠排序压不下去）。
|
||||
"""
|
||||
engine = await engines.get(source)
|
||||
|
||||
def _collect(sync_conn: Any) -> list[str]:
|
||||
def _names(sync_conn: Any) -> list[str]:
|
||||
try:
|
||||
names = sa_inspect(sync_conn).get_schema_names()
|
||||
return list(sa_inspect(sync_conn).get_schema_names())
|
||||
except Exception: # noqa: BLE001 - 有些方言不支持
|
||||
return []
|
||||
return [n for n in names if n.lower() not in _SYSTEM_SCHEMAS]
|
||||
|
||||
stats = await schema_stats(source)
|
||||
if stats:
|
||||
# 有统计就只留真有对象的，并按对象数降序——业务库自然浮到前面
|
||||
return [s for s, n in stats if n > 0 and s.lower() not in _SYSTEM_SCHEMAS]
|
||||
|
||||
async with engine.connect() as conn:
|
||||
return await conn.run_sync(_collect)
|
||||
names = await conn.run_sync(_names)
|
||||
return [n for n in names if n.lower() not in _SYSTEM_SCHEMAS]
|
||||
|
||||
|
||||
def _target_schema(source: Any, explicit: str | None) -> str | None:
|
||||
@@ -116,8 +164,18 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any
|
||||
}
|
||||
return {"tables": tables, "truncated": truncated, "total": len(names) + len(views)}
|
||||
|
||||
async with engine.connect() as conn:
|
||||
payload = await conn.run_sync(_collect)
|
||||
try:
|
||||
async with engine.connect() as conn:
|
||||
payload = await conn.run_sync(_collect)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
# schema 名填错是高频事故（大小写、拼写、或者根本不知道该填什么），
|
||||
# 直接把驱动异常抛出去等于让人自己去猜。先看看连接本身好不好：
|
||||
# 能列出候选说明只是 schema 不对，这种情况给出可选项比报错有用得多。
|
||||
candidates = await list_schemas(source)
|
||||
if not candidates:
|
||||
raise
|
||||
payload = {"tables": {}, "truncated": False, "total": 0,
|
||||
"error": f"{type(exc).__name__}: {exc}"}
|
||||
|
||||
payload["synced_at"] = datetime.now(timezone.utc).isoformat()
|
||||
payload["schema"] = target
|
||||
@@ -125,7 +183,7 @@ async def introspect(source: Any, *, schema: str | None = None) -> dict[str, Any
|
||||
# 一张表都没探到时，别只回一个空壳——多半是数据在别的 schema 里，
|
||||
# 把候选列出来，用户才知道下一步该填什么
|
||||
if not payload["tables"]:
|
||||
payload["available_schemas"] = await list_schemas(source)
|
||||
payload.setdefault("available_schemas", await list_schemas(source))
|
||||
return payload
|
||||
|
||||
|
||||
|
||||
@@ -157,3 +157,71 @@ def test_oracle_accepts_database_as_service_name():
|
||||
def test_unknown_kind_is_rejected():
|
||||
with pytest.raises(ValueError, match="不支持"):
|
||||
build_url(_src("mongodb"))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# schema 发现：各家的 "schema" 根本不是同一个东西
|
||||
# MySQL schema == database（SHOW schemas）
|
||||
# Oracle schema == user（all_users）
|
||||
# PG database > schema（pg_namespace）
|
||||
# SQLite 只有 main + attached
|
||||
# 所以这组用例守的是"不管哪一家，探不到东西时都要给出可走的下一步"。
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
async def test_sqlite_degrades_without_information_schema(source):
|
||||
"""SQLite 没有 information_schema，统计路径要安静地降级而不是报错。"""
|
||||
from app.data.introspect import list_schemas, schema_stats
|
||||
|
||||
assert await schema_stats(source) == []
|
||||
assert isinstance(await list_schemas(source), list)
|
||||
|
||||
|
||||
async def test_qualified_name_omits_schema_when_absent(source):
|
||||
"""没有 schema 概念时不要硬拼前缀，否则生成的 SQL 反而错。"""
|
||||
source.schema_cache = await introspect(source)
|
||||
assert source.schema_cache["tables"]["orders"]["qualified"] == "orders"
|
||||
|
||||
|
||||
async def test_qualified_name_includes_schema_when_given(source):
|
||||
"""有 schema 时必须带前缀——Oracle 上少了它就是 ORA-00942。"""
|
||||
cache = await introspect(source, schema="main")
|
||||
assert cache["tables"]["orders"]["qualified"] == "main.orders"
|
||||
|
||||
|
||||
async def test_bad_schema_returns_candidates_instead_of_raising(source, monkeypatch):
|
||||
"""schema 填错是高频事故（大小写、拼写），要给可选项而不是抛驱动异常。"""
|
||||
from app.data import introspect as mod
|
||||
|
||||
async def fake_list(_source):
|
||||
return ["ANALYTICS", "STAGING"]
|
||||
|
||||
monkeypatch.setattr(mod, "list_schemas", fake_list)
|
||||
cache = await mod.introspect(source, schema="根本不存在")
|
||||
assert cache["tables"] == {}
|
||||
assert cache["available_schemas"] == ["ANALYTICS", "STAGING"]
|
||||
|
||||
|
||||
async def test_summary_points_at_candidate_schemas(source, monkeypatch):
|
||||
"""摘要要把候选说给 Copilot 听，否则它只会对着空结构编表名。"""
|
||||
from app.data import introspect as mod
|
||||
|
||||
async def fake_list(_source):
|
||||
return ["ANALYTICS", "STAGING"]
|
||||
|
||||
monkeypatch.setattr(mod, "list_schemas", fake_list)
|
||||
source.schema_cache = await mod.introspect(source, schema="根本不存在")
|
||||
text = summary(source)
|
||||
assert "ANALYTICS" in text and "该账号名下没有对象" in text
|
||||
|
||||
|
||||
def test_schema_stats_sql_covers_every_supported_kind():
|
||||
"""新增数据库类型时别忘了这里——漏了就退化成黑名单猜测。"""
|
||||
from app.data.engine import SUPPORTED_KINDS
|
||||
from app.data.introspect import _SCHEMA_STATS_SQL
|
||||
|
||||
# sqlite 显式短路，其余要么有专用 SQL，要么落到 information_schema 标准路径
|
||||
assert "oracle" in _SCHEMA_STATS_SQL
|
||||
assert "_standard" in _SCHEMA_STATS_SQL
|
||||
for kind in SUPPORTED_KINDS:
|
||||
assert kind in _SCHEMA_STATS_SQL or kind in ("sqlite", "mysql", "mariadb",
|
||||
"postgres", "postgresql")
|
||||
|
||||
Reference in New Issue
Block a user