mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 04:31:58 +00:00
数据剖析改成逐表的后台任务：引擎改编排、剖析任务、接口、设置换新键
以前剖析是一次同步长请求，预算按整次算（max_queries 60、max_total_s 120），选 50 张表只做得完前两三张，其余整张记「查询次数用完」；不能停，关掉页面报告就没了，对业务库查了什么无从查。
- data/catalog_profile.py：检查逻辑和 _apply 并入规则原样保留，只换编排。公开的逐表入口 profile_table：先列全检查项（plan_checks，纯函数，范围的 profile_checks、建任务时列的检查项、执行用同一份；遮罩、多列、不在表结构里、指向视图的关系在这里就列成按规则跳过），第一条查询估行数（确知空表整张表跳过），逐项执行、进度回调，整张表结束写目录并和回调在同一次库操作里；停止时做完的项照常写、没做的记 cancelled 保留旧结论。_Runner 拆成整个任务一个的 SourceRunner（只读守卫、单条时限、任务级止损：连续 3 条失败、最近 10 条 7 条超时；最近 6 条 4 条超时提醒；每条查询留痕，SQL 经 normalize_sql 去掉字面量）和每张表一个的 TableBudget（用完只停这张表）。删掉同步的 profile_catalog、挑表前批量估行数（_skip_empty、PROFILE_DEFAULT_TABLES、PROFILE_PROBE_TABLES）和进程内 _RUNNING。设置换成 table_max_queries（1–200，缺省 40）、table_max_s（10–600，缺省 60），旧键 max_queries、max_total_s 启动时从库里删掉（migrate_settings，不折算：旧值按整次算，含义不同），保存时只摘掉不报错（without_legacy）。表助手的 run_query 和起草时核对有效记录条件改走单条查询入口 run_single_query，行为不变。
- data/catalog_profile_jobs.py：剖析任务的建（selected / needed / all 三种范围，排序同前：有待核实关系的在前、使用次数、表结构顺序；挑出 0 张 ProfileNothing）、排队（和同源起草按建立先后串行，waiting_for）、执行（每张表开始前重读数据源：停用或关掉剖析就停下；遮罩按最新设置）、停止、重试、启动标中断不续跑、保留最近 50 个结束的任务（某张表最近一次剖析所在的不删）、last_profile / profile_due 的查询、任务和逐表的接口形状。
- api/catalog.py：加 /profile-jobs（建、active、latest、{job_id}、tasks/{seq}、cancel、retry），登记在 /{table} 前面；删掉 POST /catalog/profile；清单每行加 last_profile、profile_due、profile_due_reason、profile_checks，顶层加 active_profile_job；单表加 last_profile；导入目录、同步起草遇到剖析任务 409 profile_busy。api/coded.py 加 profile_nothing、catalog_profile_finished；api/datasources.py 保存时摘掉旧键；main.py 启动时标中断加设置迁移，关停时停掉进行中的剖析。
- 测试：test_catalog_profile.py、test_catalog_profile_units.py、test_catalog_semantics.py 里打同步接口和旧设置键的改走任务和新键，检查逻辑的断言保留；同步接口的整次总量上限一条改成逐表次数、逐表时长两条。
This commit is contained in:
ThearchyHelios
+178
-81
@@ -8,12 +8,17 @@
|
||||
- POST /catalog/{table}/patch 保存目录修改提案（乐观锁），记为人工确认
|
||||
- GET /catalog/{table}/impact 影响面：引用这张表的已发布、受管模板（data/catalog_impact.py）
|
||||
- POST /catalog/draft 同步起草（注释、外键、命名推断，可选模型）。界面改用下面的后台起草任务，这个留着；
|
||||
和后台任务双向互斥：同一个源有排队或进行中的后台任务（或正在同步起草）时 409
|
||||
catalog_draft_busy，同步起草期间建任务、重试、导入也是 409
|
||||
同一个源有排队或进行中的起草任务（或正在同步起草）时 409 catalog_draft_busy，有剖析任务时
|
||||
409 profile_busy；同步起草期间建起草任务、重试、导入也是 409，新建的剖析任务排在它后面
|
||||
- PUT /catalog/guidance 数据源级的起草要求（使用者补充的背景和要求，起草时交给模型）
|
||||
- POST /catalog/draft-jobs 建后台起草任务（202）；GET /draft-jobs/active、/latest、/{job_id} 读任务；
|
||||
POST /draft-jobs/{job_id}/cancel 停止，/retry 只重试没写好的表（data/catalog_jobs.py）
|
||||
- POST /catalog/profile 同步剖析：对业务库发少量只读查询，核对关系、取码值候选（data/catalog_profile.py）
|
||||
POST /draft-jobs/{job_id}/cancel 停止，/retry 只重试没写好的表（data/catalog_jobs.py）。
|
||||
同源有剖析任务时照建，排在它后面（waiting_for）
|
||||
- POST /catalog/profile-jobs 建后台剖析任务（202）：对业务库逐表发少量只读查询，核对关系、取码值候选、提议业务日期
|
||||
（data/catalog_profile_jobs.py，引擎在 data/catalog_profile.py）；GET /profile-jobs/active、
|
||||
/latest、/{job_id} 读任务，/{job_id}/tasks/{seq} 读一张表的明细（检查项、发现、发过的查询）；
|
||||
POST /profile-jobs/{job_id}/cancel 停止，/retry 只重试没做完的表。同源有起草任务时照建，
|
||||
排在它后面（waiting_for）
|
||||
- GET /catalog/review-queue 审阅队列：推断的项按使用次数、风险、来源排好；POST /catalog/review/batch 批量审阅，
|
||||
GET /catalog/review-progress 审阅进度，GET /catalog/{table}/usage 用过这张表的运行
|
||||
（data/catalog_review.py）
|
||||
@@ -23,8 +28,9 @@
|
||||
（data/catalog_assistant.py）
|
||||
|
||||
规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径
|
||||
不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。
|
||||
起草任务的 409 带机读码（api/coded.py）：同一个源已有排队或进行中的任务、起草模型用不了、要停止的任务已经结束。
|
||||
不对 422，源或表不存在 404。所有写操作记下署名（api/deps.actor_of：开了登录是认证过的用户名，否则请求头 X-Actor 的自报）。
|
||||
起草、剖析任务的 409 带机读码（api/coded.py）：同一个源已有排队或进行中的同种任务、起草模型用不了、剖析没开启、按范围
|
||||
挑不出表、要停止的任务已经结束。
|
||||
|
||||
上传表格的源按当前快照的表结构回答（table_versions.resolve_source），和查询实际用的库是同一个版本。
|
||||
"""
|
||||
@@ -38,10 +44,10 @@ from pydantic import BaseModel, Field, ValidationError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.coded import (
|
||||
CATALOG_DRAFT_BUSY, CATALOG_DRAFT_FINISHED, CATALOG_DRAFT_MODEL_UNAVAILABLE, PROFILE_BUSY, SCHEMA_MISSING,
|
||||
SNAPSHOT_TAMPERED, CodedHTTPException,
|
||||
CATALOG_DRAFT_BUSY, CATALOG_DRAFT_FINISHED, CATALOG_DRAFT_MODEL_UNAVAILABLE, CATALOG_PROFILE_FINISHED, PROFILE_BUSY,
|
||||
PROFILE_NOTHING, SCHEMA_MISSING, SNAPSHOT_TAMPERED, CodedHTTPException,
|
||||
)
|
||||
from app.api.runs import actor_of
|
||||
from app.api.deps import actor_of
|
||||
from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, catalog_profile_jobs, introspect
|
||||
from app.data import table_versions
|
||||
from app.data import catalog_assistant, catalog_review, catalog_transfer
|
||||
@@ -103,9 +109,11 @@ GUIDANCE_MISSING = "请求里要写起草要求（guidance 一项，一段文字
|
||||
GUIDANCE_NOT_TEXT = "起草要求要是一段文字（要清除起草要求请传空串）"
|
||||
|
||||
|
||||
class CatalogProfileIn(BaseModel):
|
||||
#: 要剖析的表；不给或为空时按使用次数取前 10 张目录里有待核实关系的表
|
||||
tables: list[str] | None = Field(default=None, max_length=200)
|
||||
class ProfileJobIn(BaseModel):
|
||||
#: 要剖析的表（范围 selected，按给的顺序去重）；最多 500 张（超了 422，说明是中文）
|
||||
tables: list[str] | None = None
|
||||
#: 不指定表时的范围：needed 需要剖析的表（缺省）、all 全部表（都不含视图、没有可剖析的项的表）
|
||||
scope: Literal["needed", "all"] | None = None
|
||||
|
||||
|
||||
async def _resolved(session: AsyncSession, source_id: str) -> tuple[DataSource, Any]:
|
||||
@@ -135,7 +143,8 @@ def _item_value(notes: dict[str, Any], name: str) -> tuple[Any, str | None]:
|
||||
|
||||
|
||||
def _summary_row(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogEntry | None,
|
||||
usage: int, last_draft: dict[str, Any] | None = None) -> dict[str, Any]:
|
||||
usage: int, last_draft: dict[str, Any] | None = None,
|
||||
profiling: dict[str, Any] | None = None) -> dict[str, Any]:
|
||||
notes = entry.notes if entry else {}
|
||||
label, label_status = _item_value(notes, "label")
|
||||
kind, _ = _item_value(notes, "kind")
|
||||
@@ -158,9 +167,20 @@ def _summary_row(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogE
|
||||
# 这张表最近一次起草任务里的那一项：{status, job_id, finished_at, error, model_error, blank_columns}，没起草过
|
||||
# 是 null
|
||||
"last_draft": last_draft,
|
||||
# 剖析：最近一次剖析（CatalogLastProfile，没剖析过是 null）、要不要剖析和为什么、按当前目录和表结构要做几项检查
|
||||
**(profiling or {"last_profile": None, "profile_due": False, "profile_due_reason": None, "profile_checks": 0}),
|
||||
}
|
||||
|
||||
|
||||
def _profiling(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogEntry | None, *,
|
||||
tables: dict[str, Any], masked: set[str], last: dict[str, Any] | None) -> dict[str, Any]:
|
||||
notes = entry.notes if entry else {}
|
||||
checks = catalog_profile.profile_checks(name, meta, notes, tables=tables, masked=masked)
|
||||
reason = catalog_profile_jobs.due_reason(meta, notes, last, checks)
|
||||
return {"last_profile": catalog_profile_jobs.public_last(last), "profile_due": reason is not None,
|
||||
"profile_due_reason": reason, "profile_checks": checks}
|
||||
|
||||
|
||||
def _structure(meta: dict[str, Any]) -> dict[str, Any]:
|
||||
return {
|
||||
"qualified": meta.get("qualified"),
|
||||
@@ -189,6 +209,9 @@ async def _detail(session: AsyncSession, source: Any, table: str, entry: catalog
|
||||
"structure": _structure(meta) if meta is not None else None,
|
||||
"system_notes": catalog.system_notes_source(source),
|
||||
"usage": usage,
|
||||
# 这张表最近一次剖析（CatalogProfileTaskDetail：检查项、发现、跳过的项、日期范围、发过的查询）；任务还在进行就是
|
||||
# 进行中的状态；没剖析过是 null
|
||||
"last_profile": await catalog_profile_jobs.last_detail(session, source.id, table),
|
||||
}
|
||||
|
||||
|
||||
@@ -224,11 +247,17 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi
|
||||
entries = await catalog.read_catalog(session, source.id)
|
||||
usage = await catalog.table_usage(session, source)
|
||||
drafts = await catalog_jobs.last_drafts(session, source.id)
|
||||
lasts = await catalog_profile_jobs.last_profiles(session, source.id)
|
||||
masked = catalog_profile.masked_of(row.options)
|
||||
|
||||
def profiling(name: str, meta: dict[str, Any] | None) -> dict[str, Any]:
|
||||
return _profiling(name, meta, entries.get(name), tables=tables, masked=masked, last=lasts.get(name))
|
||||
|
||||
order = {name: i for i, name in enumerate(tables)}
|
||||
rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0), drafts.get(name))
|
||||
rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0), drafts.get(name), profiling(name, meta))
|
||||
for name, meta in tables.items()]
|
||||
rows.sort(key=lambda r: (-r["usage"], order[r["table_name"]]))
|
||||
rows += [_summary_row(name, None, entry, 0, drafts.get(name))
|
||||
rows += [_summary_row(name, None, entry, 0, drafts.get(name), profiling(name, None))
|
||||
for name, entry in sorted(entries.items()) if name not in tables]
|
||||
return {
|
||||
"tables": rows,
|
||||
@@ -239,8 +268,11 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi
|
||||
"schema_total": max(_int(cache.get("total")), len(tables)),
|
||||
# 数据源级的起草要求（没有是 null）
|
||||
"guidance": catalog_jobs.guidance_of(row),
|
||||
# 排队或进行中的起草任务 {id, status, total, done}，界面据此显示进度；没有是 null
|
||||
# 排队或进行中的起草任务 {id, status, total, done, waiting_for}，界面据此显示进度；没有是 null
|
||||
"active_job": await catalog_jobs.job_brief(session, await catalog_jobs.active_job(session, source.id)),
|
||||
# 排队或进行中的剖析任务 {id, status, total, done, waiting_for}；没有是 null
|
||||
"active_profile_job": await catalog_profile_jobs.job_brief(
|
||||
session, await catalog_profile_jobs.active_job(session, source.id)),
|
||||
}
|
||||
|
||||
|
||||
@@ -331,6 +363,136 @@ async def import_catalog(source_id: str, payload: CatalogImportIn, x_actor: str
|
||||
return report.as_dict()
|
||||
|
||||
|
||||
# ---- 后台剖析任务（同样登记在 GET /catalog/{table} 前面：active、latest 要先于 /profile-jobs/{job_id}）
|
||||
|
||||
|
||||
async def _profile_source(session: AsyncSession, source_id: str) -> tuple[DataSource, Any]:
|
||||
"""建剖析任务、重试之前的几道门，409 都带机读码（界面按它分情况给下一步，不从原话里认）：未开启 profile_disabled、
|
||||
数据源停用 datasource_inactive、快照登记被改动 snapshot_tampered、没有表结构 schema_missing。返回 (数据源记录,
|
||||
读表结构用的源)。"""
|
||||
row = await _source_row(session, source_id)
|
||||
try:
|
||||
catalog_profile.ensure_enabled(row)
|
||||
except catalog_profile.ProfileDisabled as e:
|
||||
raise CodedHTTPException(409, str(e), e.code) from e
|
||||
try:
|
||||
source = await table_versions.resolve_source(session, row)
|
||||
except SnapshotTampered as e: # 快照清单对不上（SnapshotManifestMismatch 同时是 SnapshotMissing）
|
||||
raise CodedHTTPException(409, str(e), SNAPSHOT_TAMPERED) from e
|
||||
except table_versions.SnapshotMissing as e:
|
||||
raise HTTPException(409, str(e)) from e
|
||||
cache = source.schema_cache or {}
|
||||
if not cache.get("tables"):
|
||||
raise CodedHTTPException(409, f"{introspect.why_empty(cache)}，无法进行数据剖析。请先在数据源卡片上点「探查结构」",
|
||||
SCHEMA_MISSING)
|
||||
return row, source
|
||||
|
||||
|
||||
async def _start_profile(session: AsyncSession, row: DataSource, source: Any, *, tables: list[str] | None,
|
||||
scope: str, actor: str | None, retry_of: str | None = None) -> dict[str, Any]:
|
||||
try:
|
||||
job, tasks = await catalog_profile_jobs.create_job(
|
||||
session, row, source, tables=tables, scope=scope, settings=catalog_profile.ensure_enabled(row),
|
||||
actor=actor, retry_of=retry_of)
|
||||
except catalog_profile_jobs.ProfileBusy as e:
|
||||
raise CodedHTTPException(409, str(e), PROFILE_BUSY) from e
|
||||
except catalog_profile_jobs.ProfileNothing as e:
|
||||
raise CodedHTTPException(409, str(e), PROFILE_NOTHING) from e
|
||||
catalog_profile_jobs.manager.start(job.id)
|
||||
return {"job": catalog_profile_jobs.job_dict(job, tasks,
|
||||
waiting_for=await catalog_profile_jobs.waiting_for(session, job))}
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/profile-jobs", status_code=202)
|
||||
async def create_profile_job(source_id: str, payload: ProfileJobIn, x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""建后台剖析任务，马上返回（202，任务是排队中）。tables 非空时范围是 selected（表结构里没有、视图、没有可剖析的项照样
|
||||
建进任务、状态 skipped），否则按 scope（缺省 needed）挑。同源有排队或进行中的剖析任务 409 profile_busy（说清进行到第几张；
|
||||
界面改去 GET /profile-jobs/active 显示那一个），按范围挑出 0 张表 409 profile_nothing；同源有起草任务时照建，排在它后面
|
||||
（waiting_for = "draft"），起草结束后自动开始。"""
|
||||
if payload.tables and len(payload.tables) > catalog_profile_jobs.MAX_TABLES:
|
||||
raise HTTPException(422, f"一次最多剖析 {catalog_profile_jobs.MAX_TABLES} 张表，当前 {len(payload.tables)} 张，"
|
||||
"请分几次剖析")
|
||||
row, source = await _profile_source(session, source_id)
|
||||
tables = [str(t) for t in payload.tables] if payload.tables else None
|
||||
scope = "selected" if tables else (payload.scope or "needed")
|
||||
return await _start_profile(session, row, source, tables=tables, scope=scope, actor=actor_of(x_actor))
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/profile-jobs/active")
|
||||
async def active_profile_job(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""排队或进行中的那个剖析任务，没有是 null。"""
|
||||
row = await _source_row(session, source_id)
|
||||
return {"job": await catalog_profile_jobs.job_out(session, await catalog_profile_jobs.active_job(session, row.id))}
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/profile-jobs/latest")
|
||||
async def latest_profile_job(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""最近建的一个剖析任务（任何状态），没有是 null。"""
|
||||
row = await _source_row(session, source_id)
|
||||
return {"job": await catalog_profile_jobs.job_out(session, await catalog_profile_jobs.latest_job(session, row.id))}
|
||||
|
||||
|
||||
async def _profile_job_or_404(session: AsyncSession, source_id: str, job_id: str) -> tuple[DataSource, Any]:
|
||||
row = await _source_row(session, source_id)
|
||||
job = await catalog_profile_jobs.load(session, row.id, job_id)
|
||||
if job is None:
|
||||
raise HTTPException(404, "剖析任务不存在，可能属于别的数据源，已随数据源删除，或者超过保留数被清理了")
|
||||
return row, job
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/profile-jobs/{job_id}")
|
||||
async def get_profile_job(source_id: str, job_id: str,
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
_, job = await _profile_job_or_404(session, source_id, job_id)
|
||||
return {"job": await catalog_profile_jobs.job_out(session, job)}
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/profile-jobs/{job_id}/tasks/{seq}")
|
||||
async def get_profile_task(source_id: str, job_id: str, seq: int,
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""剖析任务里的一张表（按剖析顺序的序号）：检查项、发现、跳过的项、日期范围、发过的查询（字面量已换成「?」）。"""
|
||||
_, job = await _profile_job_or_404(session, source_id, job_id)
|
||||
task = await catalog_profile_jobs.load_task(session, job, seq)
|
||||
if task is None:
|
||||
raise HTTPException(404, "这次剖析里没有这一张表")
|
||||
return {"task": catalog_profile_jobs.task_detail(job, task)}
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/profile-jobs/{job_id}/cancel")
|
||||
async def cancel_profile_job(source_id: str, job_id: str,
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""停止：排队的直接停止；进行中的不再等正在跑的那条查询（数据库按单条时限截止它），这张表做完的项照常写进目录、没做的
|
||||
保留旧结论，后面的表记停止。已经结束的 409 catalog_profile_finished。返回停止之后的任务。"""
|
||||
row, job = await _profile_job_or_404(session, source_id, job_id)
|
||||
owner = row.id
|
||||
try:
|
||||
await catalog_profile_jobs.cancel(session, job)
|
||||
except catalog_profile_jobs.ProfileFinished as e:
|
||||
raise CodedHTTPException(409, str(e), CATALOG_PROFILE_FINISHED) from e
|
||||
session.expire_all() # 后台在别的会话里收的尾：重新读
|
||||
return {"job": await catalog_profile_jobs.job_out(session, await catalog_profile_jobs.load(session, owner, job_id))}
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/profile-jobs/{job_id}/retry", status_code=202)
|
||||
async def retry_profile_job(source_id: str, job_id: str, x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""重试：新任务，只含原任务里 failed、partial、cancelled 的表（按原来的顺序），范围 selected，设置取数据源当前的。
|
||||
同源有排队或进行中的剖析任务 409 profile_busy；没有可重试的表 422；开始不了的几种情况同建任务。"""
|
||||
_, old = await _profile_job_or_404(session, source_id, job_id)
|
||||
row, source = await _profile_source(session, source_id)
|
||||
try:
|
||||
await catalog_profile_jobs.ensure_free(session, row)
|
||||
except catalog_profile_jobs.ProfileBusy as e:
|
||||
raise CodedHTTPException(409, str(e), PROFILE_BUSY) from e
|
||||
try:
|
||||
tables = await catalog_profile_jobs.retry_tables(session, old)
|
||||
except catalog_profile_jobs.NothingToRetry as e:
|
||||
raise HTTPException(422, str(e)) from e
|
||||
return await _start_profile(session, row, source, tables=tables, scope="selected", actor=actor_of(x_actor),
|
||||
retry_of=old.id)
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/{table}")
|
||||
async def get_catalog_table(source_id: str, table: str,
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
@@ -545,71 +707,6 @@ async def draft_catalog(source_id: str, payload: CatalogDraftIn,
|
||||
}
|
||||
|
||||
|
||||
def _profile_table_out(t: catalog_profile.TableProfile) -> dict[str, Any]:
|
||||
return {
|
||||
"table_name": t.table,
|
||||
"queries": t.queries,
|
||||
"row_estimate": t.row_estimate.to_dict() if t.row_estimate else None,
|
||||
"findings": t.findings,
|
||||
"skipped": [s.to_dict() for s in t.skipped],
|
||||
"date_ranges": t.date_ranges,
|
||||
"added": t.added,
|
||||
"updated": t.updated,
|
||||
"removed": t.removed,
|
||||
"version": t.version,
|
||||
"error": t.error,
|
||||
}
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/profile")
|
||||
async def profile_catalog_tables(source_id: str, payload: CatalogProfileIn,
|
||||
x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""同步剖析。返回每张表用了几条查询、写进目录的发现、跳过了什么及原因。
|
||||
|
||||
默认关闭：数据源设置里没开启数据剖析、数据源停用，409；同一个数据源已经在剖析，409。
|
||||
"""
|
||||
row, source = await _resolved(session, source_id)
|
||||
# 409 都带机读码（api/coded.py），界面按它分五种情况给下一步，不从原话里认
|
||||
try:
|
||||
catalog_profile.ensure_enabled(row)
|
||||
except catalog_profile.ProfileDisabled as e:
|
||||
raise CodedHTTPException(409, str(e), e.code) from e
|
||||
cache = source.schema_cache or {}
|
||||
if not cache.get("tables"):
|
||||
raise CodedHTTPException(409, f"{introspect.why_empty(cache)}，无法进行数据剖析。请先在数据源卡片上点「探查结构」",
|
||||
SCHEMA_MISSING)
|
||||
actor = actor_of(x_actor)
|
||||
try:
|
||||
report = await catalog_profile.profile_catalog(session, row, source, tables=payload.tables or None,
|
||||
actor=actor)
|
||||
except catalog_profile.ProfileDisabled as e:
|
||||
raise CodedHTTPException(409, str(e), e.code) from e
|
||||
except catalog_profile.ProfileBusy as e:
|
||||
raise CodedHTTPException(409, str(e), PROFILE_BUSY) from e
|
||||
except SnapshotTampered as e:
|
||||
raise CodedHTTPException(409, str(e), SNAPSHOT_TAMPERED) from e
|
||||
rows = [_profile_table_out(t) for t in report.tables]
|
||||
note = None
|
||||
if not rows and report.empty_tables:
|
||||
note = (f"有待核对关联关系的 {len(report.empty_tables)} 张表都是空表，没有可以核实的数据。"
|
||||
"请在数据有了之后再剖析，或指定要剖析的表")
|
||||
elif not rows:
|
||||
note = "目录里还没有待核实的关联关系，请先起草数据目录，或指定要剖析的表"
|
||||
return {
|
||||
"profiled_at": report.profiled_at,
|
||||
"actor": actor,
|
||||
"settings": report.settings.to_dict(),
|
||||
"queries_used": report.queries_used,
|
||||
"stopped": report.stopped,
|
||||
"tables": rows,
|
||||
"total": {k: sum(r[k] for r in rows) for k in ("added", "updated", "removed")},
|
||||
# 不指定表时挑表途中确知是空表、跳过了的表（指定了表时总是空列表）
|
||||
"empty_tables": report.empty_tables,
|
||||
"note": note,
|
||||
}
|
||||
|
||||
|
||||
@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),
|
||||
|
||||
@@ -11,14 +11,18 @@ from fastapi.responses import JSONResponse
|
||||
#: 限定的数据源一个都不在了（删了或停用了），见 copilot._sources
|
||||
DATASOURCE_SCOPE_EMPTY = "datasource_scope_empty"
|
||||
|
||||
#: 保存数据源时剖析设置不合规（422）：field 是出错的那一项（max_queries……），说不清是哪一项时没有 field
|
||||
#: 保存数据源时剖析设置不合规（422）：field 是出错的那一项（table_max_queries……），说不清是哪一项时没有 field
|
||||
PROFILE_SETTINGS_INVALID = "profile_settings_invalid"
|
||||
#: 剖析开始不了（409）的五种情况，见 api/catalog.profile_catalog_tables
|
||||
#: 剖析任务开始不了（409，api/catalog.create_profile_job）：未开启、数据源停用、没有表结构、快照被改动、这个源已有排队或
|
||||
#: 进行中的剖析任务（导入目录、同步起草遇到剖析任务也是它）、按范围挑出 0 张表
|
||||
PROFILE_DISABLED = "profile_disabled"
|
||||
DATASOURCE_INACTIVE = "datasource_inactive"
|
||||
SCHEMA_MISSING = "schema_missing"
|
||||
PROFILE_BUSY = "profile_busy"
|
||||
SNAPSHOT_TAMPERED = "snapshot_tampered"
|
||||
PROFILE_NOTHING = "profile_nothing"
|
||||
#: 要停止的剖析任务已经结束了（409）
|
||||
CATALOG_PROFILE_FINISHED = "catalog_profile_finished"
|
||||
#: 数据目录的后台起草任务（api/catalog.py 的 /catalog/draft-jobs）：同一个源已有排队或进行中的任务（409，界面改去
|
||||
#: GET /draft-jobs/active 显示那一个）、起草模型用不了（409，不建任务）、要停止的任务已经结束了（409）
|
||||
CATALOG_DRAFT_BUSY = "catalog_draft_busy"
|
||||
|
||||
@@ -23,7 +23,7 @@ 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.catalog_profile import profile_settings_rejection
|
||||
from app.data.catalog_profile import profile_settings_rejection, without_legacy
|
||||
from app.data import introspect as introspect_mod
|
||||
from app.data import table_versions
|
||||
from app.data.engine import (
|
||||
@@ -465,12 +465,13 @@ async def create_source(
|
||||
)).scalar_one_or_none()
|
||||
if exists:
|
||||
raise HTTPException(409, f"已存在名为「{payload.name}」的数据源，请换一个标识")
|
||||
_refuse_bad_options(payload.options)
|
||||
options = without_legacy(payload.options) or {}
|
||||
_refuse_bad_options(options)
|
||||
_refuse_reserved(payload.kind, payload.database)
|
||||
|
||||
row = DataSource(
|
||||
**payload.model_dump(exclude={"password", "options"}),
|
||||
options=_with_managed(payload.options, None),
|
||||
options=_with_managed(options, None),
|
||||
password=encrypt(payload.password),
|
||||
)
|
||||
_settle_oracle(row, None, None)
|
||||
@@ -519,6 +520,8 @@ async def update_source(
|
||||
# 升级前就指向上传目录的手工源：启动时已强制只读（table_versions.startup），不许再改回可写
|
||||
raise HTTPException(422, _UPLOAD_DIR_READONLY)
|
||||
if "options" in data:
|
||||
# 剖析设置里还带着以前按整次剖析算的预算（老页面提交的）：只摘掉，不报错（catalog_profile.LEGACY_KEYS）
|
||||
data["options"] = without_legacy(data["options"])
|
||||
_refuse_bad_options(data["options"])
|
||||
data["options"] = _with_managed(data["options"] or {}, row.options)
|
||||
if "password" in data:
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
是这个值又确认过的不提；有效记录条件是模型写的 SQL，另按起草的规矩核对（_Tools.filter_problem：一个条件、只用这张表的
|
||||
列、不能有子查询……）。核对过的作为 proposal 事件交给界面，**不写目录**。没收下的项和原因交回模型（不带任何取值）。
|
||||
界面上还没保存的修改（pending）里同一列的码值并进新提的码值：界面上同一路径以后一次为准，不能把先提的码值丢了。
|
||||
- run_query：只读查询，只有数据源开了数据剖析才给。经剖析的只读守卫和单条时限（catalog_profile._Runner：
|
||||
- run_query：只读查询，只有数据源开了数据剖析才给。经剖析的单条查询入口（catalog_profile.run_single_query：
|
||||
guard.check(readonly=True)，单条时限取剖析设置和数据源查询时限中较小的一个），每次请求最多 MAX_QUERIES 条（守卫拒掉的
|
||||
也算），最多 QUERY_ROWS 行交给模型、SHOWN_ROWS 行交给界面。结果交给模型、界面之前先遮（engine/masking.mask_query_result：
|
||||
取自遮罩列的结果列整列换成「已遮罩」，改了名字的也认；这一次请求里前面遮掉的值按值换），报错原话里的值也换。没开剖析时
|
||||
@@ -52,7 +52,6 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import dataclasses
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import AsyncIterator, Iterable, Mapping
|
||||
@@ -621,15 +620,11 @@ class _Tools:
|
||||
known = known if known.strong else None
|
||||
error: str | None = None
|
||||
result: Any = None
|
||||
settings = dataclasses.replace(self.ctx.settings, max_queries=1)
|
||||
try:
|
||||
runner = catalog_profile._Runner(self.ctx.query_source, settings,
|
||||
catalog_profile.SqlDialect(self.ctx.kind))
|
||||
result = await runner(sql, max_rows=QUERY_ROWS)
|
||||
except catalog_profile._Skip as e:
|
||||
result = await catalog_profile.run_single_query(self.ctx.query_source, sql, settings=self.ctx.settings,
|
||||
max_rows=QUERY_ROWS, kind=self.ctx.kind)
|
||||
except catalog_profile.QueryFailed as e:
|
||||
error = e.detail
|
||||
except catalog_profile._Stop as e:
|
||||
error = str(e)
|
||||
except SnapshotTampered as e:
|
||||
error = QUERY_TAMPERED.format(name=self.ctx.source_name, error=e)
|
||||
except ValueError as e: # 不支持的数据库类型
|
||||
|
||||
+680
-289
File diff suppressed because it is too large
Load Diff
@@ -1,20 +1,107 @@
|
||||
"""数据目录的后台剖析任务（草稿，下一步补全）。"""
|
||||
"""数据目录的后台剖析任务（/api/datasources/{id}/catalog/profile-jobs）。
|
||||
|
||||
以前剖析是一次同步长请求（POST /catalog/profile，已去掉）：预算按整次剖析算，选 50 张表只做得完前两三张；进行中只有一条
|
||||
按时长走的横条、不能停；关掉页面报告就没了，对业务库查了什么无从查。现在和起草一样是后台任务（生命周期在
|
||||
data/catalog_job_runtime.py），逐表剖析（引擎在 data/catalog_profile.py）：
|
||||
|
||||
- **建任务**（create_job）：范围三种——指定的表（selected：按给的顺序去重；表结构里没有、视图、没有可剖析的项照样建进
|
||||
任务、状态 skipped）、需要剖析的表（needed：从没剖析过、上次剖析没做完、剖析之后冒出了新的待核实关系，见 due_reason）、
|
||||
全部表（all）。后两种只挑表结构里、不是视图、有可剖析的项的表，有待核实关系的在前，再按使用次数、表结构顺序；挑出 0 张
|
||||
抛 ProfileNothing。剖析设置随任务记下快照，每张表的检查项按当时的目录先列一遍。同一个源同时只有一个排队或进行中的
|
||||
剖析任务（ProfileBusy，说法里写明进行到第几张；部分唯一索引兜底）；和同源的起草任务按建立的先后排队（每个源一条队），
|
||||
起草在前时 waiting_for 是 "draft"。建任务时顺手清掉超出保留数的旧任务（prune）。
|
||||
- **在后台剖析**（manager.start）：轮到了才开始。每张表开始前重读数据源——停用了、关掉了剖析就停下，剩下的表 cancelled、
|
||||
error 写明；遮罩按最新的设置——再逐表调 catalog_profile.profile_table（它读的是那时目录里的最新内容：起草刚提的关系也
|
||||
核对）。检查项、正在做哪一项、查询数、发过的查询随时落库；整张表的结果和写目录在同一次库操作里，停止时也不会被打断。
|
||||
- **任务级止损**（替代以前的整次总量上限）：连续 3 条查询失败、最近 10 条里 7 条超时，整个任务停下（stop_reason、error），
|
||||
正在做的那张表写入已做完的项，后面的表 cancelled、error 带同一句原因；最近 6 条里 4 条超时时 notice 有一句提醒，好转后
|
||||
清掉。任务最后的状态按逐表算：有 failed、partial、cancelled 的表是 partial，否则 succeeded；任务整体出错（数据源没了、
|
||||
快照被改动、意外异常）是 failed。
|
||||
- **停止**（cancel）：排队的直接 cancelled；进行中的停掉正在等的那条查询（库那边按单条时限截止），那张表做完的项照常写进
|
||||
目录、没做的保留旧结论，后面的表 cancelled，任务 cancelled。
|
||||
- **重试**（retry_tables + create_job）：新任务，只含原任务里 failed、partial、cancelled 的表（按原来的顺序），范围 selected，
|
||||
retry_of 记原任务；设置取数据源当前的。
|
||||
- **服务重启**（startup）：排队、进行中的标 interrupted，没结束的表 cancelled、error「服务重启，剖析中断」。不自动续跑：
|
||||
复制出来的数据目录启动时不能自己去查业务库，界面让人点「重试」。关停时（manager.shutdown）停掉进行中的，同样标中断。
|
||||
启动时还一次性迁移剖析设置的旧键（catalog_profile.migrate_settings）。
|
||||
- **保留**（prune）：每个源保留最近 KEEP_JOBS 个结束了的剖析任务，更早的删掉；某张表最近一次剖析所在的任务不删，表详情的
|
||||
「上次剖析」一直在。
|
||||
- **读**：job_dict（任务和逐表一行）、task_detail（一张表的检查项、发现、跳过的项、发过的查询）、last_profiles（表清单每行
|
||||
的「上次剖析」）、due_reason（这张表要不要剖析、为什么）。
|
||||
|
||||
没结束的表收尾时（停止、中断、整体出错）检查项里还排着的、正在做的记为跳过（原因 cancelled 或 error），不留「排队中」的
|
||||
检查项在一张已经结束的表上。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy import delete, func, select, update
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.data.catalog_job_runtime import ACTIVE, TASK_DONE
|
||||
from app.db.models import CatalogProfileJob, CatalogProfileTask
|
||||
from app.data import catalog, catalog_profile
|
||||
from app.data import catalog_job_runtime as runtime
|
||||
from app.data.catalog_job_runtime import ACTIVE, FINISHED, OPEN, RETRYABLE, STOP_GRACE_S, TASK_DONE, TASK_STATUSES
|
||||
from app.data.catalog_profile import (
|
||||
NOTHING_ERROR, UNFINISHED_REASONS, VIEW_ERROR, ProfileSettings, ProfileSkip, TableHooks, TableResult,
|
||||
)
|
||||
from app.data.engine import SnapshotTampered
|
||||
from app.db.base import SessionLocal, new_id, utcnow
|
||||
from app.db.models import CatalogDraftJob, CatalogProfileJob, CatalogProfileTask, DataSource
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: 一个任务最多几张表
|
||||
MAX_TABLES = 500
|
||||
#: 每个源保留几个结束了的剖析任务（某张表最近一次剖析所在的任务另外保留）
|
||||
KEEP_JOBS = 50
|
||||
|
||||
INTERRUPTED_ERROR = "服务重启，剖析中断"
|
||||
STOPPED_ERROR = catalog_profile.STOPPED
|
||||
UNFINISHED_ERROR = "剖析结束时这张表还没有收尾，请重试这张表"
|
||||
JOB_ERROR = "剖析任务出错：{line}"
|
||||
SOURCE_GONE = "数据源不存在，可能已被删除"
|
||||
SOURCE_DISABLED = "数据源「{name}」已停用，剖析已停止"
|
||||
PROFILE_TURNED_OFF = "数据源「{name}」已关闭数据剖析，剖析已停止"
|
||||
#: 没结束的检查项在收尾时的说明（{why} 是停止、中断、出错的原因）
|
||||
_UNCHECKED = "{why}，未检查"
|
||||
_CHECK_OPEN = ("pending", "running")
|
||||
|
||||
|
||||
class ProfileBusy(Exception):
|
||||
"""这个数据源已经有排队或进行中的剖析任务。message 给人看。"""
|
||||
"""这个数据源已经有排队或进行中的剖析任务。message 给人看（写明进行到第几张）。"""
|
||||
|
||||
|
||||
class ProfileFinished(Exception):
|
||||
"""要停止的剖析任务已经结束了。"""
|
||||
|
||||
|
||||
class ProfileNothing(Exception):
|
||||
"""按范围挑出 0 张表。message 说明原因、建议怎么办。"""
|
||||
|
||||
|
||||
class NothingToRetry(Exception):
|
||||
"""原任务里没有需要重试的表。"""
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 读任务
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
def _iso(value: Any) -> str | None:
|
||||
return value.isoformat() if value is not None else None
|
||||
|
||||
|
||||
def _empty_summary() -> dict[str, int]:
|
||||
return catalog_profile.summarize([])
|
||||
|
||||
|
||||
async def active_job(session: AsyncSession, source_id: str) -> CatalogProfileJob | None:
|
||||
"""排队或进行中的那个（同一个源最多一个）。"""
|
||||
return (await session.execute(
|
||||
select(CatalogProfileJob).where(CatalogProfileJob.source_id == source_id,
|
||||
CatalogProfileJob.status.in_(ACTIVE))
|
||||
@@ -22,16 +109,715 @@ async def active_job(session: AsyncSession, source_id: str) -> CatalogProfileJob
|
||||
)).scalar_one_or_none()
|
||||
|
||||
|
||||
def busy_message(row: Any, progress: tuple[int, int] | None) -> str:
|
||||
async def latest_job(session: AsyncSession, source_id: str) -> CatalogProfileJob | None:
|
||||
"""最近建的一个（任何状态）。"""
|
||||
return (await session.execute(
|
||||
select(CatalogProfileJob).where(CatalogProfileJob.source_id == source_id)
|
||||
.order_by(CatalogProfileJob.created_at.desc()).limit(1)
|
||||
)).scalar_one_or_none()
|
||||
|
||||
|
||||
async def load(session: AsyncSession, source_id: str, job_id: str) -> CatalogProfileJob | None:
|
||||
"""这个源的这个任务；不存在、不属于这个源都是 None。"""
|
||||
job = await session.get(CatalogProfileJob, job_id)
|
||||
return job if job is not None and job.source_id == source_id else None
|
||||
|
||||
|
||||
#: 逐表一行要的列：不读发现、跳过的项、发过的查询（任务进行中界面每秒都要取一次）
|
||||
_ROW_COLUMNS = (CatalogProfileTask.seq, CatalogProfileTask.table_name, CatalogProfileTask.status,
|
||||
CatalogProfileTask.checks, CatalogProfileTask.current, CatalogProfileTask.queries,
|
||||
CatalogProfileTask.elapsed_ms, CatalogProfileTask.row_estimate, CatalogProfileTask.summary,
|
||||
CatalogProfileTask.skipped_count, CatalogProfileTask.unfinished, CatalogProfileTask.added,
|
||||
CatalogProfileTask.updated, CatalogProfileTask.removed, CatalogProfileTask.version,
|
||||
CatalogProfileTask.error, CatalogProfileTask.started_at, CatalogProfileTask.finished_at)
|
||||
|
||||
|
||||
async def task_rows(session: AsyncSession, job_id: str) -> list[Any]:
|
||||
return list((await session.execute(
|
||||
select(*_ROW_COLUMNS).where(CatalogProfileTask.job_id == job_id).order_by(CatalogProfileTask.seq)
|
||||
)).all())
|
||||
|
||||
|
||||
def _by_kind(checks: list[Any]) -> dict[str, list[int]]:
|
||||
"""检查项按种类的进度：{relation: [结束了的, 总数], codes: […], date: […]}。结束了的含跳过的。"""
|
||||
out = {"relation": [0, 0], "codes": [0, 0], "date": [0, 0]}
|
||||
for check in checks:
|
||||
if not isinstance(check, dict) or check.get("kind") not in out:
|
||||
continue
|
||||
pair = out[check["kind"]]
|
||||
pair[1] += 1
|
||||
if check.get("state") in ("done", "skipped"):
|
||||
pair[0] += 1
|
||||
return out
|
||||
|
||||
|
||||
def task_row(task: Any) -> dict[str, Any]:
|
||||
"""逐表一行（CatalogProfileTaskRow）。task 是 CatalogProfileTask，或者 task_rows 取的那几列。"""
|
||||
checks = [c for c in (task.checks or []) if isinstance(c, dict)]
|
||||
return {
|
||||
"seq": task.seq,
|
||||
"table_name": task.table_name,
|
||||
"status": task.status,
|
||||
"checks_total": len(checks),
|
||||
"checks_done": sum(1 for c in checks if c.get("state") in ("done", "skipped")),
|
||||
"checks_by_kind": _by_kind(checks),
|
||||
"current": task.current if task.status == "running" else None,
|
||||
"queries": task.queries or 0,
|
||||
"elapsed_ms": task.elapsed_ms or 0,
|
||||
"row_estimate": task.row_estimate,
|
||||
"summary": {**_empty_summary(), **(task.summary or {})},
|
||||
"skipped_count": task.skipped_count or 0,
|
||||
"unfinished": task.unfinished or 0,
|
||||
"added": task.added or 0,
|
||||
"updated": task.updated or 0,
|
||||
"removed": task.removed or 0,
|
||||
"version": task.version or 0,
|
||||
"error": task.error,
|
||||
"started_at": _iso(task.started_at),
|
||||
"finished_at": _iso(task.finished_at),
|
||||
}
|
||||
|
||||
|
||||
def _counts(statuses: list[str]) -> dict[str, int]:
|
||||
counts = dict.fromkeys(TASK_STATUSES, 0)
|
||||
for status in statuses:
|
||||
counts[status] = counts.get(status, 0) + 1
|
||||
return counts
|
||||
|
||||
|
||||
def job_dict(job: CatalogProfileJob, tasks: list[Any], *, waiting_for: str | None = None) -> dict[str, Any]:
|
||||
"""接口返回的任务（CatalogProfileJob）：逐表按剖析顺序；计数、合计都按逐表的算。"""
|
||||
rows = [task_row(t) for t in tasks]
|
||||
counts = _counts([r["status"] for r in rows])
|
||||
summary = _empty_summary()
|
||||
for r in rows:
|
||||
for key in summary:
|
||||
summary[key] += int(r["summary"].get(key) or 0)
|
||||
return {
|
||||
"id": job.id,
|
||||
"source_id": job.source_id,
|
||||
"status": job.status,
|
||||
"scope": job.scope,
|
||||
"settings": ProfileSettings.from_dict(job.settings).to_dict(),
|
||||
"created_at": _iso(job.created_at),
|
||||
"started_at": _iso(job.started_at),
|
||||
"finished_at": _iso(job.finished_at),
|
||||
"created_by": job.created_by,
|
||||
"retry_of": job.retry_of,
|
||||
"waiting_for": waiting_for,
|
||||
"total": len(rows),
|
||||
"done": sum(counts.get(s, 0) for s in TASK_DONE),
|
||||
"counts": counts,
|
||||
"queries": job.queries or 0,
|
||||
"timeouts": job.timeouts or 0,
|
||||
"errors": job.errors or 0,
|
||||
"stop_reason": job.stop_reason,
|
||||
"notice": job.notice,
|
||||
"error": job.error,
|
||||
"added": sum(r["added"] for r in rows),
|
||||
"updated": sum(r["updated"] for r in rows),
|
||||
"removed": sum(r["removed"] for r in rows),
|
||||
"summary": summary,
|
||||
"tasks": rows,
|
||||
}
|
||||
|
||||
|
||||
async def waiting_for(session: AsyncSession, job: CatalogProfileJob) -> str | None:
|
||||
"""排队中的剖析任务在等什么：排在同源的起草任务（或同步起草）后面时是 "draft"，否则 None。
|
||||
|
||||
先看这个源的队（进程内，排在它前面的是谁）；任务刚建、后台还没排进队时看库：同源有比它先建的、排队或进行中的起草
|
||||
任务，或者正在同步起草。"""
|
||||
if job.status != "queued":
|
||||
return None
|
||||
ahead = runtime.lanes.ahead(job.source_id, job.id)
|
||||
if ahead is not None:
|
||||
return "draft" if any(kind in ("draft", "sync_draft") for kind in ahead) else None
|
||||
from app.data import catalog_jobs
|
||||
|
||||
if catalog_jobs.sync_drafting(job.source_id):
|
||||
return "draft"
|
||||
earlier = (await session.execute(
|
||||
select(CatalogDraftJob.id).where(CatalogDraftJob.source_id == job.source_id,
|
||||
CatalogDraftJob.status.in_(ACTIVE),
|
||||
CatalogDraftJob.created_at <= job.created_at).limit(1)
|
||||
)).first()
|
||||
return "draft" if earlier is not None else None
|
||||
|
||||
|
||||
async def job_out(session: AsyncSession, job: CatalogProfileJob | None) -> dict[str, Any] | None:
|
||||
if job is None:
|
||||
return None
|
||||
return job_dict(job, await task_rows(session, job.id), waiting_for=await waiting_for(session, job))
|
||||
|
||||
|
||||
async def job_brief(session: AsyncSession, job: CatalogProfileJob | None) -> dict[str, Any] | None:
|
||||
"""目录清单顶上的进度（CatalogActiveProfileJob）：{id, status, total, done, waiting_for}。"""
|
||||
if job is None:
|
||||
return None
|
||||
rows = (await session.execute(
|
||||
select(CatalogProfileTask.status, func.count()).where(CatalogProfileTask.job_id == job.id)
|
||||
.group_by(CatalogProfileTask.status)
|
||||
)).all()
|
||||
counts = {status: n for status, n in rows}
|
||||
return {"id": job.id, "status": job.status, "total": sum(counts.values()),
|
||||
"done": sum(counts.get(s, 0) for s in TASK_DONE), "waiting_for": await waiting_for(session, job)}
|
||||
|
||||
|
||||
def task_detail(job: CatalogProfileJob, task: CatalogProfileTask) -> dict[str, Any]:
|
||||
"""一张表的明细（CatalogProfileTaskDetail）：逐表一行，加上检查项、发现、跳过的项、日期范围、发过的查询。"""
|
||||
return {
|
||||
**task_row(task),
|
||||
"job_id": job.id,
|
||||
"job_status": job.status,
|
||||
"created_by": job.created_by,
|
||||
"checks": [c for c in task.checks or [] if isinstance(c, dict)],
|
||||
"findings": list(task.findings or []),
|
||||
"skipped": list(task.skipped or []),
|
||||
"date_ranges": list(task.date_ranges or []),
|
||||
"query_log": list(task.query_log or []),
|
||||
}
|
||||
|
||||
|
||||
async def load_task(session: AsyncSession, job: CatalogProfileJob, seq: int) -> CatalogProfileTask | None:
|
||||
return (await session.execute(
|
||||
select(CatalogProfileTask).where(CatalogProfileTask.job_id == job.id, CatalogProfileTask.seq == seq)
|
||||
)).scalar_one_or_none()
|
||||
|
||||
|
||||
def _latest_per_table(source_id: str, *columns: Any) -> Any:
|
||||
"""每张表最近一次剖析的那一项（最近建的任务里的；同一个任务里同名的取 seq 小的）：子查询，带 rank 列。"""
|
||||
rank = func.row_number().over(partition_by=CatalogProfileTask.table_name,
|
||||
order_by=(CatalogProfileJob.created_at.desc(), CatalogProfileTask.seq)).label("rank")
|
||||
return (select(*columns, rank).join(CatalogProfileJob, CatalogProfileJob.id == CatalogProfileTask.job_id)
|
||||
.where(CatalogProfileJob.source_id == source_id)).subquery()
|
||||
|
||||
|
||||
async def last_profiles(session: AsyncSession, source_id: str) -> dict[str, dict[str, Any]]:
|
||||
"""每张表最近一次剖析里的那一项：{表名: CatalogLastProfile 加上 paths（检查项的路径，判断有没有新的待核实关系用）}。
|
||||
任务还在进行，就是进行中的状态。每张表取哪一项在库里挑好。"""
|
||||
inner = _latest_per_table(source_id, CatalogProfileTask.table_name, CatalogProfileTask.job_id,
|
||||
CatalogProfileTask.seq, CatalogProfileTask.status, CatalogProfileTask.started_at,
|
||||
CatalogProfileTask.finished_at, CatalogProfileTask.summary,
|
||||
CatalogProfileTask.skipped_count, CatalogProfileTask.unfinished,
|
||||
CatalogProfileTask.error, CatalogProfileTask.row_estimate, CatalogProfileTask.checks)
|
||||
rows = (await session.execute(select(inner).where(inner.c.rank == 1))).mappings()
|
||||
out: dict[str, dict[str, Any]] = {}
|
||||
for row in rows:
|
||||
out[row["table_name"]] = {
|
||||
"job_id": row["job_id"], "seq": row["seq"], "status": row["status"],
|
||||
"started_at": _iso(row["started_at"]), "finished_at": _iso(row["finished_at"]),
|
||||
"summary": {**_empty_summary(), **(row["summary"] or {})}, "skipped_count": row["skipped_count"] or 0,
|
||||
"unfinished": row["unfinished"] or 0, "error": row["error"], "row_estimate": row["row_estimate"],
|
||||
"paths": {c.get("path") for c in row["checks"] or [] if isinstance(c, dict) and c.get("path")},
|
||||
}
|
||||
return out
|
||||
|
||||
|
||||
def public_last(last: dict[str, Any] | None) -> dict[str, Any] | None:
|
||||
"""接口里的 CatalogLastProfile：去掉内部用的 paths。"""
|
||||
return None if last is None else {k: v for k, v in last.items() if k != "paths"}
|
||||
|
||||
|
||||
async def last_detail(session: AsyncSession, source_id: str, table: str) -> dict[str, Any] | None:
|
||||
"""这张表最近一次剖析的明细（表详情的「数据剖析」一节）；没剖析过是 None。"""
|
||||
found = (await session.execute(
|
||||
select(CatalogProfileTask, CatalogProfileJob)
|
||||
.join(CatalogProfileJob, CatalogProfileJob.id == CatalogProfileTask.job_id)
|
||||
.where(CatalogProfileJob.source_id == source_id, CatalogProfileTask.table_name == table)
|
||||
.order_by(CatalogProfileJob.created_at.desc(), CatalogProfileTask.seq).limit(1)
|
||||
)).first()
|
||||
return None if found is None else task_detail(found[1], found[0])
|
||||
|
||||
|
||||
def due_reason(meta: dict[str, Any] | None, notes: dict[str, Any] | None, last: dict[str, Any] | None,
|
||||
checks: int) -> str | None:
|
||||
"""这张表要不要剖析、为什么：never 从没剖析过；unfinished 最近一次剖析是 partial / failed / cancelled；new_relations
|
||||
目录里有待核实的关系不在最近一次剖析的检查项里。表结构里没有、视图、没有可剖析的项（checks 为 0）、正在剖析的不要。"""
|
||||
if not isinstance(meta, dict) or meta.get("is_view") or checks <= 0:
|
||||
return None
|
||||
if last is None:
|
||||
return "never"
|
||||
if last["status"] in OPEN:
|
||||
return None
|
||||
if last["status"] in RETRYABLE:
|
||||
return "unfinished"
|
||||
if catalog_profile.pending_relation_paths(notes) - set(last.get("paths") or ()):
|
||||
return "new_relations"
|
||||
return None
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 建任务、重试、停止
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
def busy_message(row: Any, brief: dict[str, Any] | None, *, table: str | None = None,
|
||||
position: int | None = None) -> str:
|
||||
"""同源已有剖析任务时的说法：写明进行到第几张；从表详情点「剖析这张表」时再说这张表在不在那个任务里、排第几。"""
|
||||
name = getattr(row, "name", "") or ""
|
||||
return f"数据源「{name}」正在剖析，请等这次剖析结束（或先停止它）再开始新的剖析"
|
||||
if brief is None:
|
||||
text = f"数据源「{name}」正在剖析"
|
||||
elif brief["status"] == "queued":
|
||||
text = f"数据源「{name}」已有一个排队中的剖析任务（共 {brief['total']} 张表）"
|
||||
else:
|
||||
text = f"数据源「{name}」正在剖析：第 {min(brief['done'] + 1, brief['total'])} / {brief['total']} 张表"
|
||||
if table is not None:
|
||||
text += f"，{table} 在这次剖析里，排第 {position} 张" if position else f"，{table} 不在这次剖析里"
|
||||
return text + "。请等这次剖析结束（或先停止它）再开始新的剖析"
|
||||
|
||||
|
||||
async def ensure_free(session: AsyncSession, row: Any) -> None:
|
||||
async def ensure_free(session: AsyncSession, row: Any, tables: list[str] | None = None) -> None:
|
||||
"""这个源没有排队或进行中的剖析任务；有就抛 ProfileBusy。只指定了一张表时，说法里带上这张表在不在那个任务里。"""
|
||||
job = await active_job(session, row.id)
|
||||
if job is not None:
|
||||
rows = (await session.execute(
|
||||
select(CatalogProfileTask.status, func.count()).where(CatalogProfileTask.job_id == job.id)
|
||||
.group_by(CatalogProfileTask.status))).all()
|
||||
counts = dict(rows)
|
||||
raise ProfileBusy(busy_message(row, (sum(counts.get(s, 0) for s in TASK_DONE), sum(counts.values()))))
|
||||
if job is None:
|
||||
return
|
||||
brief = await job_brief(session, job)
|
||||
table = position = None
|
||||
if tables and len(tables) == 1:
|
||||
key = catalog.resolve_table_name(getattr(row, "schema_cache", None), str(tables[0])) or str(tables[0])
|
||||
table = key
|
||||
names = [r.table_name for r in await task_rows(session, job.id)]
|
||||
position = names.index(key) + 1 if key in names else None
|
||||
raise ProfileBusy(busy_message(row, brief, table=table, position=position))
|
||||
|
||||
|
||||
#: 挑出 0 张表时的说法
|
||||
NOTHING_NEEDED = ("没有需要剖析的表：能剖析的表都剖析过了，上次剖析也都做完了，之后没有新的待核实关系。"
|
||||
"要重新剖析，请选「全部表」或指定要剖析的表")
|
||||
NOTHING_AT_ALL = ("没有可以剖析的表：表结构里的表都是视图，或者没有可剖析的项（要核对的关联关系、像码值的列、日期列都没有，"
|
||||
"或者都被遮罩了）。请先起草数据目录，或指定要剖析的表")
|
||||
|
||||
|
||||
def _skipped_task(job_id: str, seq: int, name: str, reason: str, why: str, checks: list[dict[str, Any]],
|
||||
skipped: list[ProfileSkip], now: Any) -> CatalogProfileTask:
|
||||
entries = [s.to_dict() for s in skipped] + [ProfileSkip("table", name, reason, why).to_dict()]
|
||||
return CatalogProfileTask(job_id=job_id, seq=seq, table_name=name, status="skipped", checks=checks or None,
|
||||
skipped=entries, skipped_count=len(entries),
|
||||
unfinished=sum(1 for e in entries if e["reason"] in UNFINISHED_REASONS),
|
||||
summary=_empty_summary(), error=why, finished_at=now)
|
||||
|
||||
|
||||
async def _pick(session: AsyncSession, row: Any, source: Any, *, tables: list[str] | None,
|
||||
scope: str) -> list[tuple[str, str | None, str | None, list[Any]]]:
|
||||
"""要剖析的表：[(表名, 跳过的原因 / None, 整句, 检查项)]，按剖析顺序。"""
|
||||
cache = getattr(source, "schema_cache", None) or {}
|
||||
all_tables = cache.get("tables") or {}
|
||||
masked = catalog_profile.masked_of(getattr(row, "options", None))
|
||||
entries = await catalog.read_catalog(session, source.id)
|
||||
|
||||
def notes(name: str) -> dict[str, Any]:
|
||||
entry = entries.get(name)
|
||||
return entry.notes if entry else {}
|
||||
|
||||
def plan(name: str) -> list[Any]:
|
||||
return catalog_profile.plan_checks(name, all_tables.get(name), notes(name), tables=all_tables, masked=masked)
|
||||
|
||||
if scope == "selected":
|
||||
picked: list[tuple[str, str | None, str | None, list[Any]]] = []
|
||||
seen: set[str] = set()
|
||||
for name in dict.fromkeys(str(t) for t in tables or []):
|
||||
key = catalog.resolve_table_name(cache, name)
|
||||
if key is None:
|
||||
picked.append((name, "missing", catalog.missing_table_error(name), []))
|
||||
continue
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
if (all_tables.get(key) or {}).get("is_view"):
|
||||
picked.append((key, "view", VIEW_ERROR, []))
|
||||
continue
|
||||
checks = plan(key)
|
||||
if not any(c.skip is None for c in checks):
|
||||
picked.append((key, "nothing", NOTHING_ERROR, checks))
|
||||
else:
|
||||
picked.append((key, None, None, checks))
|
||||
return picked
|
||||
usage = await catalog.table_usage(session, source)
|
||||
last = await last_profiles(session, source.id) if scope == "needed" else {}
|
||||
order = {name: i for i, name in enumerate(all_tables)}
|
||||
candidates: list[tuple[str, list[Any]]] = []
|
||||
for name, meta in all_tables.items():
|
||||
if not isinstance(meta, dict) or meta.get("is_view"):
|
||||
continue
|
||||
checks = plan(name)
|
||||
runnable = sum(1 for c in checks if c.skip is None)
|
||||
if runnable == 0:
|
||||
continue
|
||||
if scope == "needed" and due_reason(meta, notes(name), last.get(name), runnable) is None:
|
||||
continue
|
||||
candidates.append((name, checks))
|
||||
if not candidates:
|
||||
raise ProfileNothing(NOTHING_NEEDED if scope == "needed" else NOTHING_AT_ALL)
|
||||
# 还有待核实（推断）关系的表排在前面：剖析首先是来核实它们的；再按使用次数、表结构顺序
|
||||
candidates.sort(key=lambda c: (not catalog_profile.has_pending_relations(notes(c[0])), -usage.get(c[0], 0),
|
||||
order[c[0]]))
|
||||
return [(name, None, None, checks) for name, checks in candidates]
|
||||
|
||||
|
||||
async def create_job(session: AsyncSession, row: Any, source: Any, *, tables: list[str] | None, scope: str,
|
||||
settings: ProfileSettings, actor: str | None,
|
||||
retry_of: str | None = None) -> tuple[CatalogProfileJob, list[CatalogProfileTask]]:
|
||||
"""建一个排队的剖析任务（还没开始，调用方接着 manager.start）。row 是数据源记录（遮罩按它的设置），source 是读表结构
|
||||
用的源（上传源是绑定当前快照的视图）。同源已有排队或进行中的剖析任务抛 ProfileBusy；挑出 0 张表抛 ProfileNothing。
|
||||
同源有起草任务时照建，排在它后面。建完顺手清掉超出保留数的旧任务。"""
|
||||
await ensure_free(session, row, tables)
|
||||
picked = await _pick(session, row, source, tables=tables, scope=scope)
|
||||
job = CatalogProfileJob(id=new_id(), source_id=row.id, status="queued", scope=scope,
|
||||
settings=settings.to_dict(), created_by=actor, retry_of=retry_of)
|
||||
now = utcnow()
|
||||
tasks: list[CatalogProfileTask] = []
|
||||
for seq, (name, reason, why, checks) in enumerate(picked):
|
||||
shown = [c.as_check() for c in checks]
|
||||
if reason is not None:
|
||||
rule = [ProfileSkip(c.kind, c.target, c.skip[0], c.skip[1], c.path) for c in checks if c.skip]
|
||||
tasks.append(_skipped_task(job.id, seq, name, reason, str(why), shown, rule, now))
|
||||
continue
|
||||
tasks.append(CatalogProfileTask(job_id=job.id, seq=seq, table_name=name, status="pending", checks=shown,
|
||||
summary=_empty_summary()))
|
||||
session.add(job)
|
||||
try:
|
||||
await session.flush()
|
||||
session.add_all(tasks)
|
||||
await session.commit()
|
||||
except IntegrityError as e:
|
||||
# 部分唯一索引：另一个请求同时给这个源建了剖析任务
|
||||
await session.rollback()
|
||||
raise ProfileBusy(busy_message(row, None)) from e
|
||||
try:
|
||||
await prune(session, row.id)
|
||||
except Exception: # noqa: BLE001 - 清旧任务失败不影响这次剖析
|
||||
logger.exception("清理数据源 %s 的旧剖析任务时出错", row.id)
|
||||
await session.rollback()
|
||||
return job, tasks
|
||||
|
||||
|
||||
async def retry_tables(session: AsyncSession, job: CatalogProfileJob) -> list[str]:
|
||||
"""重试要带上的表：原任务里 failed、partial、cancelled 的，按原来的顺序。没有抛 NothingToRetry。"""
|
||||
names = [t.table_name for t in await task_rows(session, job.id) if t.status in RETRYABLE]
|
||||
if not names:
|
||||
raise NothingToRetry("这次剖析里没有失败、部分完成或停止了的表，不需要重试")
|
||||
return names
|
||||
|
||||
|
||||
async def prune(session: AsyncSession, source_id: str, *, keep: int | None = None) -> int:
|
||||
"""保留：这个源最近 keep（缺省 KEEP_JOBS）个结束了的剖析任务留着，更早的删掉；某张表最近一次剖析所在的任务不删。
|
||||
返回删了几个。"""
|
||||
keep = KEEP_JOBS if keep is None else keep
|
||||
finished = list((await session.execute(
|
||||
select(CatalogProfileJob.id).where(CatalogProfileJob.source_id == source_id,
|
||||
CatalogProfileJob.status.in_(FINISHED))
|
||||
.order_by(CatalogProfileJob.created_at.desc(), CatalogProfileJob.id.desc())
|
||||
)).scalars())
|
||||
old = finished[keep:]
|
||||
if not old:
|
||||
return 0
|
||||
inner = _latest_per_table(source_id, CatalogProfileTask.job_id)
|
||||
protected = set((await session.execute(select(inner.c.job_id).where(inner.c.rank == 1))).scalars())
|
||||
victims = [job_id for job_id in old if job_id not in protected]
|
||||
if not victims:
|
||||
return 0
|
||||
await session.execute(delete(CatalogProfileTask).where(CatalogProfileTask.job_id.in_(victims)))
|
||||
await session.execute(delete(CatalogProfileJob).where(CatalogProfileJob.id.in_(victims)))
|
||||
await session.commit()
|
||||
return len(victims)
|
||||
|
||||
|
||||
async def _close_tasks(session: AsyncSession, job_id: str, task_status: str, task_error: str | None,
|
||||
*, why: str) -> None:
|
||||
"""还没结束的表记 task_status（原因 task_error）；它们的检查项里还排着的、正在做的记为跳过（停止、中断是 cancelled，
|
||||
出错是 error，说明是「{why}，未检查」），计进跳过的项和没做完的项。"""
|
||||
now = utcnow()
|
||||
reason = "cancelled" if task_status == "cancelled" else "error"
|
||||
detail = _UNCHECKED.format(why=why)
|
||||
tasks = (await session.execute(
|
||||
select(CatalogProfileTask).where(CatalogProfileTask.job_id == job_id, CatalogProfileTask.status.in_(OPEN))
|
||||
)).scalars()
|
||||
for task in tasks:
|
||||
checks = [dict(c) for c in task.checks or [] if isinstance(c, dict)]
|
||||
skipped = list(task.skipped or [])
|
||||
for check in checks:
|
||||
if check.get("state") in _CHECK_OPEN:
|
||||
check.update(state="skipped", reason=reason, detail=detail)
|
||||
skipped.append(ProfileSkip(check.get("kind") or "", check.get("target") or "", reason, detail,
|
||||
check.get("path")).to_dict())
|
||||
task.status, task.error, task.finished_at, task.current = task_status, task_error, now, None
|
||||
task.checks, task.skipped = checks or None, skipped or None
|
||||
task.skipped_count = len(skipped)
|
||||
task.unfinished = sum(1 for s in skipped if s.get("reason") in UNFINISHED_REASONS)
|
||||
|
||||
|
||||
async def _close(session: AsyncSession, job_id: str, job_status: str, job_error: str | None, task_status: str,
|
||||
task_error: str | None, *, why: str | None = None) -> None:
|
||||
"""收尾：还没结束的表记 task_status（见 _close_tasks），任务（还没结束的话）记 job_status。提交。"""
|
||||
await _close_tasks(session, job_id, task_status, task_error, why=why or task_error or STOPPED_ERROR)
|
||||
await session.execute(
|
||||
update(CatalogProfileJob).where(CatalogProfileJob.id == job_id, CatalogProfileJob.status.in_(ACTIVE))
|
||||
.values(status=job_status, error=job_error, finished_at=utcnow()).execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def cancel(session: AsyncSession, job: CatalogProfileJob) -> None:
|
||||
"""停止一个剖析任务。排队的直接 cancelled（在队里等着的后台执行跟着停掉、出队）；进行中的交给 manager 停：正在等的那条
|
||||
查询不再等（库那边按单条时限截止），那张表做完的项照常写进目录，后面的表 cancelled。已经结束的抛 ProfileFinished。"""
|
||||
if job.status in FINISHED:
|
||||
raise ProfileFinished("这次剖析已经结束了，不需要停止")
|
||||
if job.status == "queued":
|
||||
claimed = await session.execute(
|
||||
update(CatalogProfileJob).where(CatalogProfileJob.id == job.id, CatalogProfileJob.status == "queued")
|
||||
.values(status="cancelled", finished_at=utcnow()).execution_options(synchronize_session=False))
|
||||
if claimed.rowcount == 1:
|
||||
await _close_tasks(session, job.id, "cancelled", None, why=STOPPED_ERROR)
|
||||
await session.commit()
|
||||
await manager.stop(job.id, "user", grace=STOP_GRACE_S)
|
||||
return
|
||||
await session.rollback() # 刚被后台认领、开始了：按进行中的停
|
||||
if await manager.stop(job.id, "user", grace=STOP_GRACE_S):
|
||||
return
|
||||
# 库里说在进行，这个进程里却没有在跑它（记录对不上）：直接记停止
|
||||
await _close(session, job.id, "cancelled", None, "cancelled", None)
|
||||
|
||||
|
||||
async def startup() -> int:
|
||||
"""服务启动时：排队、进行中的剖析任务标成 interrupted，没结束的表记 cancelled、写明服务重启，不自动续跑；再把剖析设置的
|
||||
旧键迁移掉。返回标了几个任务。"""
|
||||
async def interrupt(session: AsyncSession, job_id: str) -> None:
|
||||
await _close(session, job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR)
|
||||
|
||||
n = await runtime.interrupt_on_startup(CatalogProfileJob, interrupt, running=manager.running)
|
||||
if n:
|
||||
logger.warning("服务重启：%d 个数据剖析任务没有做完，已标为中断，可在数据目录页重试", n)
|
||||
async with SessionLocal() as session:
|
||||
migrated = await catalog_profile.migrate_settings(session)
|
||||
if migrated:
|
||||
logger.info("数据剖析设置：%d 个数据源去掉了按整次剖析算的旧预算", migrated)
|
||||
return n
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 在后台剖析
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
class _ProfileRun(runtime.JobRun):
|
||||
"""一个剖析任务在后台的一次执行（进队、停止、收尾的框架在 runtime.JobRun）。"""
|
||||
|
||||
kind = "profile"
|
||||
model = CatalogProfileJob
|
||||
|
||||
def __init__(self, job_id: str) -> None:
|
||||
super().__init__(job_id)
|
||||
self.actor: str | None = None
|
||||
self.cx: catalog_profile.ProfileContext | None = None
|
||||
#: 正在剖析的那张表：seq 和开始的时刻（进度回调据此落库）
|
||||
self._seq = -1
|
||||
self._table_started = 0.0
|
||||
|
||||
# ---------------------------------------------------------------- 停止、出错时收尾
|
||||
|
||||
async def _stopped(self) -> None:
|
||||
if self.stop_reason == "user":
|
||||
await self._end("cancelled", None, "cancelled", None)
|
||||
else:
|
||||
await self._end("interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR)
|
||||
|
||||
async def _failed(self, reason: str) -> None:
|
||||
await self._end("failed", reason, "failed", reason)
|
||||
|
||||
async def _crashed(self, e: Exception) -> None:
|
||||
from app.core.errors import describe_exception, first_line
|
||||
|
||||
logger.exception("数据剖析任务 %s 出错", self.job_id)
|
||||
reason = JOB_ERROR.format(line=first_line(e) or describe_exception(e))
|
||||
await self._end("failed", reason, "failed", reason)
|
||||
|
||||
async def _end(self, job_status: str, job_error: str | None, task_status: str, task_error: str | None) -> None:
|
||||
"""停止、出错时收尾：等手上的库写入写完，记下查询计数，再把没结束的表和任务记下来。收尾本身写不进去只记日志（下次
|
||||
启动时 startup 会把它标成中断）。"""
|
||||
try:
|
||||
await self._settle()
|
||||
await self._write(self._save_counters)
|
||||
await self._write(_close, self.job_id, job_status, job_error, task_status, task_error)
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("数据剖析任务 %s 收尾时写库出错", self.job_id)
|
||||
|
||||
def stop_text(self) -> str:
|
||||
return STOPPED_ERROR if self.stop_reason == "user" else INTERRUPTED_ERROR
|
||||
|
||||
# ---------------------------------------------------------------- 主流程
|
||||
|
||||
async def _run(self) -> None:
|
||||
if not await self._write(self._claim):
|
||||
return # 开始之前就被停止了
|
||||
async with SessionLocal() as session:
|
||||
job = await session.get(CatalogProfileJob, self.job_id)
|
||||
row = await session.get(DataSource, job.source_id) if job is not None else None
|
||||
if job is None or row is None:
|
||||
raise runtime.JobFailed(SOURCE_GONE)
|
||||
from app.data import table_versions
|
||||
|
||||
try:
|
||||
source = await table_versions.resolve_source(session, row)
|
||||
except table_versions.SnapshotMissing as e:
|
||||
raise runtime.JobFailed(str(e)) from e
|
||||
pending = [(t.seq, t.table_name) for t in await task_rows(session, job.id) if t.status == "pending"]
|
||||
settings = ProfileSettings.from_dict(job.settings)
|
||||
self.actor = job.created_by
|
||||
options = dict(row.options or {})
|
||||
try:
|
||||
self.cx = catalog_profile.make_context(source, settings, options=options)
|
||||
except ValueError as e: # 不支持的数据库类型
|
||||
raise runtime.JobFailed(str(e)) from e
|
||||
hooks = TableHooks(progress=self._progress, query=self._queried, stop_text=self.stop_text)
|
||||
for seq, name in pending:
|
||||
# 每张表开始前重读数据源：停用了、关掉了剖析就停下；遮罩按最新的设置
|
||||
stop = await self._write(self._recheck)
|
||||
if stop is not None:
|
||||
await self._write(self._halt, stop, None)
|
||||
break
|
||||
self._table_started = time.monotonic()
|
||||
await self._write(self._start_table, seq)
|
||||
|
||||
async def recorded(session: AsyncSession, result: TableResult, seq: int = seq) -> None:
|
||||
await self._save_table(session, seq, result)
|
||||
|
||||
hooks.recorded = recorded
|
||||
self._seq = seq
|
||||
try:
|
||||
await catalog_profile.profile_table(self.cx, name, db=self._write, hooks=hooks, actor=self.actor)
|
||||
except SnapshotTampered as e:
|
||||
raise runtime.JobFailed(str(e)) from e
|
||||
if self.cx.runner.stopped:
|
||||
# 任务级止损：后面的表 cancelled，带同一句原因
|
||||
await self._write(self._halt, self.cx.runner.error, self.cx.runner.stopped)
|
||||
break
|
||||
await self._settle()
|
||||
await self._write(self._finish)
|
||||
|
||||
# ---------------------------------------------------------------- 进度回调（引擎调）
|
||||
|
||||
async def _progress(self, result: TableResult) -> None:
|
||||
await self._write(self._save_progress, self._seq, result)
|
||||
|
||||
async def _queried(self, result: TableResult) -> None:
|
||||
await self._write(self._save_queries, self._seq, result)
|
||||
|
||||
# ---------------------------------------------------------------- 库操作
|
||||
|
||||
def _task(self, *where: Any) -> Any:
|
||||
return update(CatalogProfileTask).where(CatalogProfileTask.job_id == self.job_id, *where) \
|
||||
.execution_options(synchronize_session=False)
|
||||
|
||||
def _counters(self) -> dict[str, Any]:
|
||||
runner = self.cx.runner if self.cx is not None else None
|
||||
if runner is None:
|
||||
return {}
|
||||
return {"queries": runner.queries, "timeouts": runner.timeouts, "errors": runner.errors,
|
||||
"notice": runner.notice}
|
||||
|
||||
async def _save_counters(self, session: AsyncSession) -> None:
|
||||
if values := self._counters():
|
||||
await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id).values(**values)
|
||||
.execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
async def _claim(self, session: AsyncSession) -> bool:
|
||||
return await runtime.claim(session, CatalogProfileJob, self.job_id)
|
||||
|
||||
async def _recheck(self, session: AsyncSession) -> str | None:
|
||||
"""重读数据源：停用了、关掉了剖析返回停下的整句；遮罩换成最新的设置。数据源没了抛 JobFailed。"""
|
||||
job = await session.get(CatalogProfileJob, self.job_id)
|
||||
row = await session.get(DataSource, job.source_id) if job is not None else None
|
||||
if row is None:
|
||||
raise runtime.JobFailed(SOURCE_GONE)
|
||||
if row.enabled is False:
|
||||
return SOURCE_DISABLED.format(name=row.name)
|
||||
if not catalog_profile.profile_settings(row.options).enabled:
|
||||
return PROFILE_TURNED_OFF.format(name=row.name)
|
||||
assert self.cx is not None
|
||||
self.cx.masked = catalog_profile.masked_of(row.options)
|
||||
return None
|
||||
|
||||
async def _halt(self, session: AsyncSession, error: str | None, stop_reason: str | None) -> None:
|
||||
"""整个任务停下（止损、数据源停用或关掉了剖析）：后面的表 cancelled、error 是同一句原因；任务记下原因，状态到收尾时
|
||||
按逐表算。"""
|
||||
await _close_tasks(session, self.job_id, "cancelled", error, why=error or STOPPED_ERROR)
|
||||
await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id)
|
||||
.values(error=error, stop_reason=stop_reason, **self._counters())
|
||||
.execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
async def _start_table(self, session: AsyncSession, seq: int) -> None:
|
||||
await session.execute(self._task(CatalogProfileTask.seq == seq, CatalogProfileTask.status == "pending")
|
||||
.values(status="running", started_at=utcnow()))
|
||||
await session.commit()
|
||||
|
||||
def _elapsed(self) -> int:
|
||||
return int((time.monotonic() - self._table_started) * 1000)
|
||||
|
||||
async def _save_progress(self, session: AsyncSession, seq: int, result: TableResult) -> None:
|
||||
"""一项检查开始或结束：这张表的检查项、正在做哪一项、查询数、发过的查询、行数估算，任务的查询计数。"""
|
||||
await session.execute(self._task(CatalogProfileTask.seq == seq, CatalogProfileTask.status == "running")
|
||||
.values(checks=[dict(c) for c in result.checks] or None, current=result.current,
|
||||
queries=result.queries, elapsed_ms=self._elapsed(),
|
||||
query_log=list(result.query_log) or None,
|
||||
row_estimate=result.row_estimate.to_dict() if result.row_estimate else None))
|
||||
if values := self._counters():
|
||||
await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id).values(**values)
|
||||
.execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
async def _save_queries(self, session: AsyncSession, seq: int, result: TableResult) -> None:
|
||||
"""发出一条查询之后：这张表和任务的查询计数、提醒（发过的查询本身在检查项结束时一起写）。"""
|
||||
await session.execute(self._task(CatalogProfileTask.seq == seq, CatalogProfileTask.status == "running")
|
||||
.values(queries=result.queries, elapsed_ms=self._elapsed()))
|
||||
if values := self._counters():
|
||||
await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id).values(**values)
|
||||
.execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
async def _save_table(self, session: AsyncSession, seq: int, result: TableResult) -> None:
|
||||
"""一张表结束（和写目录在同一次库操作里）：整张表的结果。"""
|
||||
await session.execute(
|
||||
self._task(CatalogProfileTask.seq == seq, CatalogProfileTask.status.in_(OPEN))
|
||||
.values(status=result.status, checks=[dict(c) for c in result.checks] or None, current=None,
|
||||
queries=result.queries, elapsed_ms=result.elapsed_ms,
|
||||
row_estimate=result.row_estimate.to_dict() if result.row_estimate else None,
|
||||
findings=list(result.findings) or None, skipped=[s.to_dict() for s in result.skipped] or None,
|
||||
date_ranges=list(result.date_ranges) or None, query_log=list(result.query_log) or None,
|
||||
summary=result.summary(), skipped_count=result.skipped_count, unfinished=result.unfinished,
|
||||
added=result.added, updated=result.updated, removed=result.removed, version=result.version,
|
||||
error=result.error, finished_at=utcnow()))
|
||||
if values := self._counters():
|
||||
await session.execute(update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id).values(**values)
|
||||
.execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
async def _finish(self, session: AsyncSession) -> None:
|
||||
"""剖析完：万一还有没收尾的表（进度回调写库失败之类）记失败；有 failed、partial、cancelled 的表是 partial，否则
|
||||
succeeded。"""
|
||||
await _close_tasks(session, self.job_id, "failed", UNFINISHED_ERROR, why=UNFINISHED_ERROR)
|
||||
statuses = list((await session.execute(
|
||||
select(CatalogProfileTask.status).where(CatalogProfileTask.job_id == self.job_id))).scalars())
|
||||
status = "partial" if any(s in RETRYABLE for s in statuses) else "succeeded"
|
||||
await session.execute(
|
||||
update(CatalogProfileJob).where(CatalogProfileJob.id == self.job_id, CatalogProfileJob.status == "running")
|
||||
.values(status=status, finished_at=utcnow(), **self._counters())
|
||||
.execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
|
||||
class ProfileJobs(runtime.JobManager):
|
||||
"""进程内的剖析任务（start / stop / wait / running / shutdown 见 runtime.JobManager）。"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(_ProfileRun, label="数据剖析任务", name="catalog-profile")
|
||||
|
||||
|
||||
manager = ProfileJobs()
|
||||
|
||||
|
||||
__all__ = ["INTERRUPTED_ERROR", "KEEP_JOBS", "MAX_TABLES", "NOTHING_AT_ALL", "NOTHING_NEEDED", "NothingToRetry",
|
||||
"ProfileBusy", "ProfileFinished", "ProfileJobs", "ProfileNothing", "active_job", "busy_message", "cancel",
|
||||
"create_job", "due_reason", "ensure_free", "job_brief", "job_dict", "job_out", "last_detail",
|
||||
"last_profiles", "latest_job", "load", "load_task", "manager", "prune", "public_last", "retry_tables",
|
||||
"startup", "task_detail", "task_row", "task_rows", "waiting_for"]
|
||||
|
||||
+9
-4
@@ -93,14 +93,18 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("上传表格的版本整理失败，服务照常启动")
|
||||
|
||||
# 上次没做完的数据目录起草任务：标成中断（没结束的表记停止），不自动续跑——拷来的数据目录（检查脚本、验收副本）
|
||||
# 启动时不能替人去调模型。界面上点「重试」接着起草没写好的表
|
||||
from app.data import catalog_jobs
|
||||
# 上次没做完的数据目录起草、剖析任务：标成中断（没结束的表记停止），不自动续跑——拷来的数据目录（检查脚本、验收副本）
|
||||
# 启动时不能替人去调模型、查业务库。界面上点「重试」接着做没做完的表。剖析设置里按整次剖析算的旧预算顺带迁移掉
|
||||
from app.data import catalog_jobs, catalog_profile_jobs
|
||||
|
||||
try:
|
||||
await catalog_jobs.startup()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("数据目录起草任务的启动整理失败，服务照常启动")
|
||||
try:
|
||||
await catalog_profile_jobs.startup()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("数据剖析任务的启动整理失败，服务照常启动")
|
||||
|
||||
await run_manager.setup()
|
||||
# 流程：存量的待审批补建待办（派给任何人，幂等）；上次进程没了时已发出、没拿到结果的写入收成结果未知、安排核实
|
||||
@@ -129,9 +133,10 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
# 先停调度，再停运行：停运行的途中不该再有新运行被发起。起草任务停下来标成中断，下次启动不会再续
|
||||
# 先停调度，再停运行：停运行的途中不该再有新运行被发起。起草、剖析任务停下来标成中断，下次启动不会再续
|
||||
await scheduler.stop()
|
||||
await catalog_jobs.manager.shutdown()
|
||||
await catalog_profile_jobs.manager.shutdown()
|
||||
await run_manager.shutdown()
|
||||
from app.sandbox.manager import sandbox_manager
|
||||
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
"""数据剖析接口：POST /api/datasources/{id}/catalog/profile（阶段 2A）。
|
||||
"""数据剖析：POST /api/datasources/{id}/catalog/profile-jobs 建后台任务，逐表剖析（阶段 2A，后台任务改造之后）。
|
||||
|
||||
剖析是数据目录里唯一对业务库发查询的地方，这里逐条钉住方案定的约束：默认关闭、只读、每条查询都过守卫、
|
||||
查询次数和单条时限、总时长、遮罩的列不取样、大表不做整表统计、同一个数据源同时只有一次剖析；
|
||||
每张表的查询次数和时长、单条时限、遮罩的列不取样、大表不做整表统计、同一个数据源同时只有一个剖析任务；
|
||||
以及剖析的结论怎么进目录：命名推断对上数据的升为有确证（来源记为数据剖析，和外键分开）、对不上的写明
|
||||
覆盖率、人工确认过的只补覆盖率和基数、状态列得到码值候选、只有一个日期列的表提议业务日期。
|
||||
任务本身（排队、停止、重启、重试、止损、范围、留痕、保留、设置迁移）在 test_catalog_profile_jobs.py。
|
||||
|
||||
所有测试都在合成的景区库上做（tests/fixtures/catalog/scenic.py），数据和名称全部虚构。
|
||||
"""
|
||||
@@ -22,7 +23,7 @@ from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import event
|
||||
from sqlalchemy.engine import Engine
|
||||
|
||||
from app.data import catalog, catalog_profile
|
||||
from app.data import catalog_profile, catalog_profile_jobs
|
||||
from app.data import engine as data_engine
|
||||
from app.data.catalog_profile import PROFILE_OPTION, ProfileSettings
|
||||
from app.db.base import SessionLocal
|
||||
@@ -65,6 +66,8 @@ async def make_source(client):
|
||||
return sid
|
||||
|
||||
yield _make
|
||||
# 后台任务先收完（停不下来的停掉），再删源：删源会级联删掉任务
|
||||
await catalog_profile_jobs.manager.shutdown(grace=5)
|
||||
for sid in made:
|
||||
await drop_source(sid)
|
||||
|
||||
@@ -119,7 +122,39 @@ def spy(monkeypatch, scenic_db):
|
||||
|
||||
|
||||
def _url(sid: str) -> str:
|
||||
return f"/api/datasources/{sid}/catalog/profile"
|
||||
return f"/api/datasources/{sid}/catalog/profile-jobs"
|
||||
|
||||
|
||||
async def _start(client, sid: str, tables: list[str] | None = None, **body):
|
||||
payload = {**({"tables": tables} if tables is not None else {}), **body}
|
||||
return await client.post(_url(sid), json=payload, headers=ACTOR)
|
||||
|
||||
|
||||
async def _job(client, sid: str, job_id: str) -> dict:
|
||||
r = await client.get(f"{_url(sid)}/{job_id}")
|
||||
assert r.status_code == 200, r.text
|
||||
return r.json()["job"]
|
||||
|
||||
|
||||
async def _detail(client, sid: str, job_id: str, seq: int) -> dict:
|
||||
r = await client.get(f"{_url(sid)}/{job_id}/tasks/{seq}")
|
||||
assert r.status_code == 200, r.text
|
||||
return r.json()["task"]
|
||||
|
||||
|
||||
async def _finished(client, sid: str, job_id: str) -> dict:
|
||||
"""等任务做完，返回做完的任务；job["tables"] 是每张表的明细（检查项、发现、跳过的项、发过的查询），按剖析顺序。"""
|
||||
await catalog_profile_jobs.manager.wait(job_id, timeout=30)
|
||||
job = await _job(client, sid, job_id)
|
||||
job["tables"] = [await _detail(client, sid, job_id, t["seq"]) for t in job["tasks"]]
|
||||
return job
|
||||
|
||||
|
||||
async def _profile(client, sid: str, tables: list[str] | None = None, **body) -> dict:
|
||||
"""建剖析任务（202）、等它做完，返回做完的任务（见 _finished）。"""
|
||||
r = await _start(client, sid, tables, **body)
|
||||
assert r.status_code == 202, r.text
|
||||
return await _finished(client, sid, r.json()["job"]["id"])
|
||||
|
||||
|
||||
async def _notes(client, sid: str, table: str) -> dict:
|
||||
@@ -147,31 +182,35 @@ async def test_profile_is_off_by_default_and_says_where_to_turn_it_on(client, ma
|
||||
sid = await make_source(scenic_db, draft=["visits"])
|
||||
watch = spy(scenic_db)
|
||||
watch.clear()
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
|
||||
r = await _start(client, sid, ["visits"])
|
||||
assert r.status_code == 409
|
||||
detail = r.json()["detail"]
|
||||
assert "数据源设置" in detail and "开启数据剖析" in detail
|
||||
assert r.json()["code"] == "profile_disabled" # 界面按机读码认是哪一种，不从原话里认
|
||||
assert watch.executed == [] # 一条查询都没发
|
||||
assert (await client.get(f"{_url(sid)}/latest")).json()["job"] is None, "没建任务"
|
||||
|
||||
|
||||
async def test_settings_are_saved_and_validated_through_datasource_update(client, make_source, scenic_db):
|
||||
sid = await make_source(scenic_db)
|
||||
r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(max_queries=0)})
|
||||
assert r.status_code == 422 and "查询次数上限" in r.json()["detail"]
|
||||
r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(table_max_queries=0)})
|
||||
assert r.status_code == 422 and "每张表最多查询数" in r.json()["detail"]
|
||||
# 机读码和出错的那一项：表单按 field 把报错落到那一格
|
||||
assert (r.json()["code"], r.json()["field"]) == ("profile_settings_invalid", "max_queries")
|
||||
assert (r.json()["code"], r.json()["field"]) == ("profile_settings_invalid", "table_max_queries")
|
||||
r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(table_max_s=5)})
|
||||
assert r.status_code == 422 and (r.json()["code"], r.json()["field"]) == ("profile_settings_invalid", "table_max_s")
|
||||
assert "每张表时长上限" in r.json()["detail"]
|
||||
r = await client.patch(f"/api/datasources/{sid}", json={"options": {PROFILE_OPTION: {"enabled": "是"}}})
|
||||
assert r.status_code == 422 and (r.json()["code"], r.json()["field"]) == ("profile_settings_invalid", "enabled")
|
||||
r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(rows=5)})
|
||||
assert r.status_code == 422 and r.json()["code"] == "profile_settings_invalid" and "field" not in r.json()
|
||||
r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(max_queries=40, sample_size=500)})
|
||||
r = await client.patch(f"/api/datasources/{sid}", json={"options": _on(table_max_queries=30, sample_size=500)})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["options"][PROFILE_OPTION] == {"enabled": True, "max_queries": 40, "sample_size": 500}
|
||||
# 设好以后就能剖析；返回里带着这次实际用的设置
|
||||
r = await client.post(_url(sid), json={"tables": ["channels"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["settings"]["max_queries"] == 40 and r.json()["settings"]["sample_size"] == 500
|
||||
assert r.json()["options"][PROFILE_OPTION] == {"enabled": True, "table_max_queries": 30, "sample_size": 500}
|
||||
# 设好以后就能剖析；任务里记着这次实际用的设置（快照）
|
||||
job = await _profile(client, sid, ["channels"])
|
||||
assert job["settings"] == {"enabled": True, "table_max_queries": 30, "table_max_s": 60, "query_timeout_s": 10,
|
||||
"sample_size": 500, "max_scan_rows": 100_000}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- 剖析的结论
|
||||
@@ -182,15 +221,15 @@ async def test_profile_verifies_name_relations_and_proposes_codes_and_business_d
|
||||
sid = await make_source(scenic_db, options=_on(), draft=["visits", "channels"])
|
||||
watch = spy(scenic_db)
|
||||
watch.clear()
|
||||
r = await client.post(_url(sid), json={"tables": ["visits", "channels"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
body = await _profile(client, sid, ["visits", "channels"])
|
||||
|
||||
# 所有发到业务库的语句都经过了守卫
|
||||
# 所有发到业务库的语句都经过了守卫；任务记的查询数就是发到库的条数，也是逐表的合计
|
||||
assert watch.all_guarded(), watch.executed
|
||||
assert body["queries_used"] == len(watch.executed) <= body["settings"]["max_queries"]
|
||||
assert body["queries_used"] == sum(t["queries"] for t in body["tables"])
|
||||
assert body["stopped"] is None and body["actor"] == "周宁" and body["profiled_at"]
|
||||
assert body["queries"] == len(watch.executed) == sum(t["queries"] for t in body["tables"])
|
||||
assert all(t["queries"] <= body["settings"]["table_max_queries"] for t in body["tables"])
|
||||
assert (body["status"], body["stop_reason"], body["error"]) == ("succeeded", None, None)
|
||||
assert body["created_by"] == "周宁" and body["finished_at"] and body["scope"] == "selected"
|
||||
assert [t["table_name"] for t in body["tasks"]] == ["visits", "channels"], "按给的顺序"
|
||||
|
||||
notes = (await _notes(client, sid, "visits"))
|
||||
assert notes["updated_by"] == "周宁"
|
||||
@@ -226,6 +265,7 @@ async def test_profile_verifies_name_relations_and_proposes_codes_and_business_d
|
||||
assert "channel_code" not in channels.get("columns", {}) # 唯一约束列不是码值
|
||||
|
||||
visits = _table(body, "visits")
|
||||
assert visits["status"] == "ok" and visits["unfinished"] == 0
|
||||
kinds = sorted(f["kind"] for f in visits["findings"])
|
||||
# 两条命名推断的关系（gate_id、member_id），加两条外键关系（park_id、ticket_type_id）：外键的只核实基数
|
||||
assert kinds == ["business_date", "codes", "relation", "relation", "relation", "relation"]
|
||||
@@ -243,6 +283,19 @@ async def test_profile_verifies_name_relations_and_proposes_codes_and_business_d
|
||||
assert visits["row_estimate"] == {"rows": 1500, "method": "count", "at_least": None}
|
||||
# 更新的 4 项：两条命名关系升为有确证，两条外键关系补上用数据核实过的基数
|
||||
assert visits["added"] == 2 and visits["updated"] == 4 and visits["error"] is None
|
||||
# 发现的合计：逐表、整个任务各一份
|
||||
assert visits["summary"] == {"relations_checked": 4, "relations_verified": 4, "codes": 1,
|
||||
"codes_pending": blank, "business_dates": 1}
|
||||
assert body["summary"]["relations_checked"] == 4 and body["summary"]["codes"] == 2
|
||||
# 检查项：每项都结束了，按种类的进度对得上
|
||||
assert visits["checks_done"] == visits["checks_total"] == len(visits["checks"])
|
||||
assert visits["checks_by_kind"]["relation"] == [4, 4] and visits["checks_by_kind"]["date"] == [1, 1]
|
||||
assert {c["state"] for c in visits["checks"]} == {"done"}
|
||||
assert sum(c["queries"] for c in visits["checks"]) + sum(
|
||||
1 for q in visits["query_log"] if q["kind"] == "row_estimate" and q["target"] == "visits") >= 1
|
||||
# 第一条查询是估行数
|
||||
assert visits["query_log"][0]["kind"] == "row_estimate" and visits["query_log"][0]["outcome"] == "ok"
|
||||
assert len(visits["query_log"]) == visits["queries"]
|
||||
|
||||
|
||||
async def test_wrong_name_relation_stays_proposed_with_coverage_in_note(client, make_source, tmp_path):
|
||||
@@ -261,16 +314,15 @@ async def test_wrong_name_relation_stays_proposed_with_coverage_in_note(client,
|
||||
before = _relation((await _notes(client, sid, "parking_records"))["notes"], "lot_id")
|
||||
assert (before["to_table"], before["source"], before["status"]) == ("lots", "name", "proposed")
|
||||
|
||||
r = await client.post(_url(sid), json={"tables": ["parking_records"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
body = await _profile(client, sid, ["parking_records"])
|
||||
rel = _relation((await _notes(client, sid, "parking_records"))["notes"], "lot_id")
|
||||
assert rel["status"] == "proposed" and rel["source"] == "profile"
|
||||
assert rel["coverage"] == 0.15
|
||||
assert "抽样覆盖率 15%" in rel["note"] and "可能不是这条关系" in rel["note"] and "20 个" in rel["note"]
|
||||
finding = _table(r.json(), "parking_records")["findings"][0]
|
||||
finding = _table(body, "parking_records")["findings"][0]
|
||||
assert finding["status"] == "proposed" and finding["coverage"] == 0.15
|
||||
# 两个日期类列：不提议业务日期，取值范围只在返回里
|
||||
ranges = {d["column"] for d in _table(r.json(), "parking_records")["date_ranges"]}
|
||||
# 两个日期类列：不提议业务日期，取值范围只在剖析结果里
|
||||
ranges = {d["column"] for d in _table(body, "parking_records")["date_ranges"]}
|
||||
assert ranges == {"entered_at"} # exited_at 全是空值
|
||||
assert "business_date" not in (await _notes(client, sid, "parking_records"))["notes"]
|
||||
|
||||
@@ -284,22 +336,21 @@ async def test_confirmed_relation_only_gets_coverage_and_cardinality(client, mak
|
||||
"if_version": detail["version"]})
|
||||
assert r.status_code == 200, r.text
|
||||
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
body = await _profile(client, sid, ["visits"])
|
||||
rel = _relation((await _notes(client, sid, "visits"))["notes"], "gate_id")
|
||||
assert (rel["source"], rel["status"]) == ("name", "confirmed")
|
||||
assert rel["coverage"] == 1.0 and rel["cardinality"] == "many_to_one"
|
||||
# 人定的基数不改；原来就有基数（命名推断记的多对一），剖析也不给它记「用数据核实过」
|
||||
assert "cardinality_checked" not in rel
|
||||
finding = next(f for f in _table(r.json(), "visits")["findings"] if f.get("columns") == ["gate_id"])
|
||||
finding = next(f for f in _table(body, "visits")["findings"] if f.get("columns") == ["gate_id"])
|
||||
assert finding["status"] == "confirmed" and finding["confirmed"] is True
|
||||
|
||||
|
||||
async def test_rerun_with_same_data_changes_nothing(client, make_source, scenic_db):
|
||||
"""同一天、数据没变：第二次剖析不改目录、不升版本（冻结进快照的目录不会无故变）。"""
|
||||
sid = await make_source(scenic_db, options=_on(), draft=["visits"])
|
||||
first = (await client.post(_url(sid), json={"tables": ["visits"]})).json()
|
||||
second = (await client.post(_url(sid), json={"tables": ["visits"]})).json()
|
||||
first = await _profile(client, sid, ["visits"])
|
||||
second = await _profile(client, sid, ["visits"])
|
||||
a, b = _table(first, "visits"), _table(second, "visits")
|
||||
assert (b["added"], b["updated"], b["removed"]) == (0, 0, 0)
|
||||
assert b["version"] == a["version"]
|
||||
@@ -318,32 +369,38 @@ async def test_masked_columns_are_never_queried(client, make_source, scenic_db,
|
||||
draft=["visits", "channels"])
|
||||
watch = spy(scenic_db)
|
||||
watch.clear()
|
||||
r = await client.post(_url(sid), json={"tables": ["visits", "channels"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
body = await _profile(client, sid, ["visits", "channels"])
|
||||
assert watch.all_guarded()
|
||||
assert watch.mentions("member_id") == [] and watch.mentions("channel_type") == []
|
||||
visits, channels = _table(r.json(), "visits"), _table(r.json(), "channels")
|
||||
visits, channels = _table(body, "visits"), _table(body, "channels")
|
||||
masked = [s for s in visits["skipped"] + channels["skipped"] if s["reason"] == "masked"]
|
||||
assert {s["target"] for s in masked} == {"member_id → members.id", "channel_type"}
|
||||
assert all("遮罩" in s["detail"] for s in masked)
|
||||
# 按规则不查的不算没做完：表照样是完成
|
||||
assert visits["status"] == "ok" and visits["unfinished"] == 0 and visits["skipped_count"] == len(visits["skipped"])
|
||||
# 留痕里也没有遮罩的列
|
||||
assert not any("member_id" in q["sql"].lower() or "channel_type" in q["sql"].lower()
|
||||
for t in body["tables"] for q in t["query_log"])
|
||||
notes = (await _notes(client, sid, "visits"))["notes"]
|
||||
assert _relation(notes, "member_id")["source"] == "name" # 没动
|
||||
assert _relation(notes, "gate_id")["status"] == "verified" # 其余照常
|
||||
assert "channel_type" not in ((await _notes(client, sid, "channels"))["notes"].get("columns") or {})
|
||||
|
||||
|
||||
async def test_query_budget_stops_profiling_and_reports_the_rest(client, make_source, scenic_db, spy):
|
||||
sid = await make_source(scenic_db, options=_on(max_queries=3), draft=["visits"])
|
||||
async def test_each_table_has_its_own_query_budget(client, make_source, scenic_db, spy):
|
||||
"""逐表预算：visits 用完自己的 3 条查询记部分完成，剩下的项记「这张表的查询次数用完」、旧结论原样留着；后面的
|
||||
channels 照样拿满额，做完。"""
|
||||
sid = await make_source(scenic_db, options=_on(table_max_queries=3), draft=["visits"])
|
||||
watch = spy(scenic_db)
|
||||
watch.clear()
|
||||
r = await client.post(_url(sid), json={"tables": ["visits", "channels"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
assert body["queries_used"] == 3 == len(watch.executed) and body["stopped"] == "budget"
|
||||
body = await _profile(client, sid, ["visits", "channels"])
|
||||
assert body["queries"] == len(watch.executed) and body["stop_reason"] is None
|
||||
visits, channels = _table(body, "visits"), _table(body, "channels")
|
||||
assert "budget" in _reasons(visits)
|
||||
assert channels["queries"] == 0 and _reasons(channels) == {"budget"}
|
||||
assert all("查询次数" in s["detail"] for s in visits["skipped"] + channels["skipped"] if s["reason"] == "budget")
|
||||
assert visits["status"] == "partial" and visits["queries"] == 3
|
||||
assert "budget" in _reasons(visits) and visits["unfinished"] >= 1
|
||||
assert all("这张表的查询次数" in s["detail"] for s in visits["skipped"] if s["reason"] == "budget")
|
||||
assert channels["status"] == "ok" and 0 < channels["queries"] <= 3 and "budget" not in _reasons(channels)
|
||||
assert body["status"] == "partial", "有部分完成的表"
|
||||
# 没查完的关系不改：原来的命名推断原样留着
|
||||
rel = _relation((await _notes(client, sid, "visits"))["notes"], "member_id")
|
||||
assert (rel["source"], rel["status"]) == ("name", "proposed")
|
||||
@@ -362,18 +419,20 @@ async def test_single_query_timeout_is_recorded_and_profiling_continues(client,
|
||||
return await real(source, sql, limits=limits)
|
||||
|
||||
monkeypatch.setattr(data_engine, "run_query", slow_gate)
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
visits = _table(r.json(), "visits")
|
||||
body = await _profile(client, sid, ["visits"])
|
||||
visits = _table(body, "visits")
|
||||
timeout = [s for s in visits["skipped"] if s["reason"] == "timeout"]
|
||||
assert [s["target"] for s in timeout] == ["gate_id → gates.id"] and "超时" in timeout[0]["detail"]
|
||||
assert r.json()["stopped"] is None
|
||||
assert body["stop_reason"] is None and body["timeouts"] == 1
|
||||
assert visits["status"] == "partial", "超时算运行原因没查成"
|
||||
assert [q["outcome"] for q in visits["query_log"]].count("timeout") == 1
|
||||
notes = (await _notes(client, sid, "visits"))["notes"]
|
||||
assert _relation(notes, "gate_id")["source"] == "name" # 超时的不改
|
||||
assert _relation(notes, "member_id")["status"] == "verified" # 其余照常
|
||||
|
||||
|
||||
async def test_total_deadline_stops_profiling(make_source, scenic_db, monkeypatch):
|
||||
async def test_each_table_has_its_own_deadline(make_source, scenic_db, monkeypatch):
|
||||
"""逐表时长：每条查询慢 0.5 秒，visits 的时长（1.4 秒）用完只停它自己，channels 另拿一份满额的时长。"""
|
||||
sid = await make_source(scenic_db, draft=["visits"])
|
||||
real = data_engine.run_query
|
||||
|
||||
@@ -384,13 +443,13 @@ async def test_total_deadline_stops_profiling(make_source, scenic_db, monkeypatc
|
||||
monkeypatch.setattr(data_engine, "run_query", slow)
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(DataSource, sid)
|
||||
settings = ProfileSettings(enabled=True, max_total_s=1.4)
|
||||
report = await catalog_profile.profile_catalog(session, row, row, tables=["visits"], actor=None,
|
||||
settings=settings)
|
||||
assert report.stopped == "deadline"
|
||||
assert 1 <= report.queries_used <= 3
|
||||
reasons = {s.reason for t in report.tables for s in t.skipped}
|
||||
assert "deadline" in reasons
|
||||
cx = catalog_profile.make_context(row, ProfileSettings(enabled=True, table_max_s=1.4), options=row.options)
|
||||
visits = await catalog_profile.profile_table(cx, "visits")
|
||||
assert visits.status == "partial" and 1 <= visits.queries <= 3
|
||||
assert "deadline" in {s.reason for s in visits.skipped}
|
||||
assert any("这张表的时长" in s.detail for s in visits.skipped if s.reason == "deadline")
|
||||
channels = await catalog_profile.profile_table(cx, "channels")
|
||||
assert channels.status == "ok" and channels.queries == 2, [s.to_dict() for s in channels.skipped]
|
||||
|
||||
|
||||
async def test_large_tables_get_no_full_scans(client, make_source, scenic_db, spy):
|
||||
@@ -398,15 +457,15 @@ async def test_large_tables_get_no_full_scans(client, make_source, scenic_db, sp
|
||||
sid = await make_source(scenic_db, options=_on(max_scan_rows=100), draft=["visits"])
|
||||
watch = spy(scenic_db)
|
||||
watch.clear()
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
visits = _table(r.json(), "visits")
|
||||
body = await _profile(client, sid, ["visits"])
|
||||
visits = _table(body, "visits")
|
||||
assert visits["row_estimate"] == {"rows": None, "method": "count", "at_least": 101}
|
||||
on_visits = [s for s in watch.executed if "from visits" in s.lower()]
|
||||
assert on_visits and all("LIMIT" in s for s in on_visits), on_visits
|
||||
assert not any("MIN(" in s for s in on_visits)
|
||||
assert "too_large" in _reasons(visits)
|
||||
assert any("太大" in s["detail"] or "上限" in s["detail"] for s in visits["skipped"] if s["reason"] == "too_large")
|
||||
assert visits["status"] == "ok", "表太大是按规则不查，不算没做完"
|
||||
# 父表有主键，覆盖率又够：照样有确证；子表一侧没核对唯一，按多对一记，不算用数据核实过基数
|
||||
rel = _relation((await _notes(client, sid, "visits"))["notes"], "gate_id")
|
||||
assert rel["status"] == "verified" and rel["cardinality"] == "many_to_one"
|
||||
@@ -416,7 +475,7 @@ async def test_large_tables_get_no_full_scans(client, make_source, scenic_db, sp
|
||||
# ---------------------------------------------------------------- 接口
|
||||
|
||||
|
||||
async def test_concurrent_profile_on_same_source_returns_409(client, make_source, scenic_db, monkeypatch):
|
||||
async def test_second_profile_job_on_same_source_returns_409(client, make_source, scenic_db, monkeypatch):
|
||||
sid = await make_source(scenic_db, options=_on(), draft=["visits"])
|
||||
real = data_engine.run_query
|
||||
started, release = asyncio.Event(), asyncio.Event()
|
||||
@@ -427,29 +486,39 @@ async def test_concurrent_profile_on_same_source_returns_409(client, make_source
|
||||
return await real(source, sql, limits=limits)
|
||||
|
||||
monkeypatch.setattr(data_engine, "run_query", held)
|
||||
first = asyncio.create_task(client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR))
|
||||
first = await _start(client, sid, ["visits", "channels"])
|
||||
assert first.status_code == 202, first.text
|
||||
await asyncio.wait_for(started.wait(), 10)
|
||||
second = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
|
||||
assert second.status_code == 409 and "正在进行数据剖析" in second.json()["detail"]
|
||||
assert second.json()["code"] == "profile_busy"
|
||||
second = await _start(client, sid, ["visits"])
|
||||
assert second.status_code == 409 and second.json()["code"] == "profile_busy"
|
||||
# 说清是谁在跑、跑到哪；只点了一张表时说这张表在不在那个任务里、排第几
|
||||
assert "正在剖析：第 1 / 2 张表" in second.json()["detail"] and "visits 在这次剖析里，排第 1 张" in second.json()["detail"]
|
||||
other = await _start(client, sid, ["gates"])
|
||||
assert other.status_code == 409 and "gates 不在这次剖析里" in other.json()["detail"]
|
||||
active = (await client.get(f"{_url(sid)}/active")).json()["job"]
|
||||
assert active["id"] == first.json()["job"]["id"] and active["status"] == "running"
|
||||
listed = (await client.get(f"/api/datasources/{sid}/catalog")).json()
|
||||
assert listed["active_profile_job"] == {"id": active["id"], "status": "running", "total": 2, "done": 0,
|
||||
"waiting_for": None}
|
||||
release.set()
|
||||
done = await asyncio.wait_for(first, 30)
|
||||
assert done.status_code == 200, done.text
|
||||
await catalog_profile_jobs.manager.wait(active["id"], timeout=30)
|
||||
assert (await client.get(f"{_url(sid)}/active")).json()["job"] is None
|
||||
# 做完以后可以再来
|
||||
again = await client.post(_url(sid), json={"tables": ["channels"]}, headers=ACTOR)
|
||||
assert again.status_code == 200, again.text
|
||||
again = await _start(client, sid, ["channels"])
|
||||
assert again.status_code == 202, again.text
|
||||
await catalog_profile_jobs.manager.wait(again.json()["job"]["id"], timeout=30)
|
||||
|
||||
|
||||
async def test_blocked_profile_says_which_case_by_code(client, make_source, scenic_db, monkeypatch):
|
||||
"""剖析开始不了的 409 都带机读码：数据源停用、没有表结构、快照被改动（未开启、正在剖析见上面两条）。"""
|
||||
from app.data.engine import SnapshotTampered
|
||||
"""剖析开始不了的 409 都带机读码：数据源停用、没有表结构、快照登记被改动（未开启、正忙见上面两条）。"""
|
||||
from app.data import table_versions
|
||||
|
||||
sid = await make_source(scenic_db, options=_on(), draft=["visits"])
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(DataSource, sid)
|
||||
row.enabled = False
|
||||
await session.commit()
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
|
||||
r = await _start(client, sid, ["visits"])
|
||||
assert r.status_code == 409 and r.json()["code"] == "datasource_inactive" and "已停用" in r.json()["detail"]
|
||||
|
||||
async with SessionLocal() as session:
|
||||
@@ -457,103 +526,50 @@ async def test_blocked_profile_says_which_case_by_code(client, make_source, scen
|
||||
row.enabled = True
|
||||
row.schema_cache = {}
|
||||
await session.commit()
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
|
||||
r = await _start(client, sid, ["visits"])
|
||||
assert r.status_code == 409 and r.json()["code"] == "schema_missing" and "探查结构" in r.json()["detail"]
|
||||
|
||||
r = await client.post(f"/api/datasources/{sid}/introspect")
|
||||
assert r.status_code == 200, r.text
|
||||
|
||||
async def tampered(*_args, **_kwargs):
|
||||
raise table_versions.SnapshotManifestMismatch("数据版本的哈希登记与快照清单不一致，已拒绝查询")
|
||||
|
||||
with monkeypatch.context() as m:
|
||||
m.setattr(table_versions, "resolve_source", tampered)
|
||||
r = await _start(client, sid, ["visits"])
|
||||
assert r.status_code == 409 and r.json()["code"] == "snapshot_tampered"
|
||||
assert (await client.get(f"{_url(sid)}/latest")).json()["job"] is None, "开始不了的都没建任务"
|
||||
|
||||
|
||||
async def test_a_tampered_snapshot_during_profiling_fails_the_job(client, make_source, scenic_db, monkeypatch):
|
||||
"""剖析途中数据文件和登记的版本对不上：整个任务停下、记失败，原因照实写；没做完的表也记失败，可以重试。"""
|
||||
from app.data.engine import SnapshotTampered
|
||||
|
||||
sid = await make_source(scenic_db, options=_on(), draft=["visits"])
|
||||
|
||||
async def tampered(*_args, **_kwargs):
|
||||
raise SnapshotTampered("数据文件与登记的版本不一致，可能被修改过，已拒绝查询")
|
||||
|
||||
monkeypatch.setattr(catalog_profile, "profile_catalog", tampered)
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]}, headers=ACTOR)
|
||||
assert r.status_code == 409 and r.json()["code"] == "snapshot_tampered"
|
||||
|
||||
|
||||
async def test_default_tables_are_used_tables_with_relation_candidates(client, make_source, scenic_db):
|
||||
sid = await make_source(scenic_db, options=_on(), draft=[])
|
||||
r = await client.post(_url(sid), json={}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
names = [t["table_name"] for t in r.json()["tables"]]
|
||||
assert names and len(names) <= catalog_profile.PROFILE_DEFAULT_TABLES
|
||||
async with SessionLocal() as session:
|
||||
entries = await catalog.read_catalog(session, sid)
|
||||
for name in names:
|
||||
assert catalog_profile._relation_candidates(entries[name].notes), name
|
||||
assert "parks" not in names and "channels" not in names # 没有待核实关系的表不选
|
||||
|
||||
|
||||
async def _candidates(sid: str) -> tuple[list[str], set[str]]:
|
||||
"""默认挑表的候选（目录里有要核对的关系的表，按表结构顺序）和其中还有待核实（推断）关系的那些。"""
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(DataSource, sid)
|
||||
entries = await catalog.read_catalog(session, sid)
|
||||
order = list((row.schema_cache or {}).get("tables") or {})
|
||||
rels = {name: catalog_profile._relation_candidates(entries[name].notes) for name in order if name in entries}
|
||||
names = [name for name in order if rels.get(name)]
|
||||
return names, {name for name in names if any(r["status"] == "proposed" for r in rels[name])}
|
||||
|
||||
|
||||
async def test_default_pick_skips_empty_tables_and_prefers_pending_relations(client, make_source, tmp_path,
|
||||
monkeypatch):
|
||||
"""B1：表都没被运行查询过（使用次数都是 0）时，以前按表结构顺序取前几张，结果大半是空表（景区库里排在前面的
|
||||
候选表正好多数没有数据）。现在跳过行数估计为 0 的表（记在 empty_tables 里），并且先挑还有待核实关系的表，
|
||||
不把刚核实过的再剖析一遍。默认张数调成 3，好在一个小库上看出先后。"""
|
||||
monkeypatch.setattr(catalog_profile, "PROFILE_DEFAULT_TABLES", 3)
|
||||
path = scenic.build(tmp_path / "b1.db")
|
||||
db = sqlite3.connect(path)
|
||||
try:
|
||||
tables = [r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type = 'table' ORDER BY rowid")]
|
||||
empty = {t for t in tables if db.execute(f'SELECT COUNT(*) FROM "{t}"').fetchone()[0] == 0}
|
||||
finally:
|
||||
db.close()
|
||||
sid = await make_source(str(path), options=_on(), draft=tables)
|
||||
names, pending = await _candidates(sid)
|
||||
full = [name for name in names if name not in empty]
|
||||
# 前提：按表结构顺序的前 3 张候选里有空表（以前就会挑中它们），有数据、还待核实的候选不止 6 张
|
||||
assert set(names[:3]) & empty
|
||||
assert len([name for name in full if name in pending]) >= 6
|
||||
|
||||
first = (await client.post(_url(sid), json={}, headers=ACTOR)).json()
|
||||
picked = [t["table_name"] for t in first["tables"]]
|
||||
assert len(picked) == 3 and not set(picked) & empty, picked
|
||||
assert first["empty_tables"] and set(first["empty_tables"]) <= empty
|
||||
assert all(t["queries"] > 0 for t in first["tables"])
|
||||
assert first["queries_used"] >= sum(t["queries"] for t in first["tables"]) + len(first["empty_tables"])
|
||||
|
||||
# 第二次：先挑还有待核实关系的表；第一次核实过的（关系都升为有确证了）排到后面，这次挑不到
|
||||
names, pending = await _candidates(sid)
|
||||
assert not set(picked) & pending # 第一次挑的三张，关系都核实了
|
||||
second = (await client.post(_url(sid), json={}, headers=ACTOR)).json()
|
||||
again = [t["table_name"] for t in second["tables"]]
|
||||
assert len(again) == 3 and all(name in pending for name in again), (again, pending)
|
||||
assert not set(again) & set(picked) and not set(again) & empty
|
||||
|
||||
|
||||
async def test_profile_note_when_every_candidate_is_empty(client, make_source, tmp_path):
|
||||
path = scenic.build(tmp_path / "b1-empty.db")
|
||||
sid = await make_source(str(path), options=_on(), draft=[])
|
||||
names, _ = await _candidates(sid)
|
||||
db = sqlite3.connect(path)
|
||||
try:
|
||||
for name in names:
|
||||
db.execute(f'DELETE FROM "{name}"')
|
||||
db.commit()
|
||||
finally:
|
||||
db.close()
|
||||
body = (await client.post(_url(sid), json={}, headers=ACTOR)).json()
|
||||
assert body["tables"] == [] and set(body["empty_tables"]) == set(names)
|
||||
assert "空表" in body["note"]
|
||||
monkeypatch.setattr(data_engine, "run_query", tampered)
|
||||
body = await _profile(client, sid, ["visits", "channels"])
|
||||
assert body["status"] == "failed" and body["error"] == "数据文件与登记的版本不一致，可能被修改过，已拒绝查询"
|
||||
assert {t["status"] for t in body["tables"]} == {"failed"}
|
||||
assert all(t["error"] == body["error"] for t in body["tables"])
|
||||
# 没结束的检查项不留在「排队」：记为跳过，原因写清
|
||||
assert all(c["state"] != "pending" for t in body["tables"] for c in t["checks"])
|
||||
|
||||
|
||||
async def test_unknown_table_and_missing_source(client, make_source, scenic_db):
|
||||
sid = await make_source(scenic_db, options=_on(), draft=["visits"])
|
||||
r = await client.post(_url(sid), json={"tables": ["nope"]}, headers=ACTOR)
|
||||
assert r.status_code == 200
|
||||
assert _table(r.json(), "nope")["error"] and r.json()["queries_used"] == 0
|
||||
body = await _profile(client, sid, ["nope"])
|
||||
nope = _table(body, "nope")
|
||||
assert nope["status"] == "skipped" and nope["error"] == "表结构里没有 nope，请先重新探查结构"
|
||||
assert body["queries"] == 0 and body["status"] == "succeeded"
|
||||
r = await client.post(_url("0" * 32), json={}, headers=ACTOR)
|
||||
assert r.status_code == 404
|
||||
assert (await client.get(f"{_url(sid)}/not-a-job")).status_code == 404
|
||||
assert (await client.get(f"{_url(sid)}/{body['id']}/tasks/9")).status_code == 404
|
||||
|
||||
|
||||
def _orders_db(path, *, items_per_order: int) -> str:
|
||||
@@ -583,27 +599,23 @@ async def test_foreign_key_relations_get_their_cardinality_checked_with_data(cli
|
||||
before = _relation((await _notes(client, sid, "order_items"))["notes"], "order_id")
|
||||
assert (before["source"], before["status"]) == ("fk", "verified")
|
||||
|
||||
r = await client.post(_url(sid), json={"tables": ["order_items"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
body = await _profile(client, sid, ["order_items"])
|
||||
rel = _relation((await _notes(client, sid, "order_items"))["notes"], "order_id")
|
||||
assert (rel["source"], rel["status"]) == ("fk", "verified")
|
||||
assert rel["cardinality"] == cardinality and rel.get("cardinality_checked") is True
|
||||
assert rel.get("coverage") is None
|
||||
finding = next(f for f in _table(r.json(), "order_items")["findings"] if f.get("columns") == ["order_id"])
|
||||
finding = next(f for f in _table(body, "order_items")["findings"] if f.get("columns") == ["order_id"])
|
||||
assert finding["cardinality_checked"] is True and finding["cardinality"] == cardinality
|
||||
|
||||
# 重新剖析、数据没变：版本不升
|
||||
version = (await _notes(client, sid, "order_items"))["version"]
|
||||
r = await client.post(_url(sid), json={"tables": ["order_items"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
await _profile(client, sid, ["order_items"])
|
||||
assert (await _notes(client, sid, "order_items"))["version"] == version
|
||||
|
||||
|
||||
async def test_fanout_on_a_data_checked_foreign_key_is_an_error_again(client, make_source, tmp_path):
|
||||
"""剖析核实过的外键一对多：订单关联明细后对订单金额求和，扇出检查报 error；数据上一对一的不报。"""
|
||||
from app.data.sqlcheck import load_checkers
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import DataSource
|
||||
|
||||
sql = "SELECT SUM(o.total_amount) AS gmv FROM orders o JOIN order_items i ON i.order_id = o.id"
|
||||
for items, expected in ((3, "error"), (1, None)):
|
||||
@@ -615,8 +627,7 @@ async def test_fanout_on_a_data_checked_foreign_key_is_an_error_again(client, ma
|
||||
r = await client.put(f"/api/datasources/{sid}/catalog/orders",
|
||||
json={"notes": notes, "if_version": detail["version"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
r = await client.post(_url(sid), json={"tables": ["order_items"]}, headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
await _profile(client, sid, ["order_items"])
|
||||
async with SessionLocal() as session:
|
||||
source = await session.get(DataSource, sid)
|
||||
checker = (await load_checkers(session, [source]))[source.name]
|
||||
|
||||
@@ -20,26 +20,30 @@ AT = "2026-10-03T08:00:00+00:00"
|
||||
def test_settings_default_off_and_parse_leniently():
|
||||
s = profile_settings({})
|
||||
assert s == ProfileSettings()
|
||||
assert s.enabled is False and s.max_queries == 60 and s.query_timeout_s == 10 and s.sample_size == 2000
|
||||
assert s.max_scan_rows == 100_000 and s.max_total_s == 120
|
||||
on = profile_settings({PROFILE_OPTION: {"enabled": True, "max_queries": "30", "sample_size": 500}})
|
||||
assert on.enabled is True and on.max_queries == 30 and on.sample_size == 500
|
||||
# 库里存着的值不对（老数据、手改过）：那一项用缺省，不因此开启，也不报错
|
||||
odd = profile_settings({PROFILE_OPTION: {"enabled": "yes", "max_queries": 0, "query_timeout_s": "x"}})
|
||||
assert odd.enabled is False and odd.max_queries == 60 and odd.query_timeout_s == 10
|
||||
assert s.enabled is False and s.table_max_queries == 40 and s.table_max_s == 60 and s.query_timeout_s == 10
|
||||
assert s.sample_size == 2000 and s.max_scan_rows == 100_000
|
||||
on = profile_settings({PROFILE_OPTION: {"enabled": True, "table_max_queries": "30", "sample_size": 500}})
|
||||
assert on.enabled is True and on.table_max_queries == 30 and on.sample_size == 500
|
||||
# 库里存着的值不对（老数据、手改过）：那一项用缺省，不因此开启，也不报错；迁移前的旧键不看
|
||||
odd = profile_settings({PROFILE_OPTION: {"enabled": "yes", "table_max_queries": 0, "query_timeout_s": "x",
|
||||
"max_queries": 5}})
|
||||
assert odd.enabled is False and odd.table_max_queries == 40 and odd.query_timeout_s == 10
|
||||
assert profile_settings({PROFILE_OPTION: "on"}) == ProfileSettings()
|
||||
# 任务里存着的快照读回来是同一份设置
|
||||
assert ProfileSettings.from_dict(on.to_dict()) == on and ProfileSettings.from_dict(None) == ProfileSettings()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value, label", [
|
||||
({"max_queries": 0}, "查询次数上限"),
|
||||
({"max_queries": 501}, "查询次数上限"),
|
||||
({"max_queries": 2.5}, "查询次数上限"),
|
||||
({"table_max_queries": 0}, "每张表最多查询数"),
|
||||
({"table_max_queries": 201}, "每张表最多查询数"),
|
||||
({"table_max_queries": 2.5}, "每张表最多查询数"),
|
||||
({"table_max_s": 9}, "每张表时长上限"),
|
||||
({"table_max_s": 601}, "每张表时长上限"),
|
||||
({"query_timeout_s": 61}, "单条查询时限"),
|
||||
({"sample_size": 5}, "抽样键值数"),
|
||||
({"max_scan_rows": -1}, "整表统计行数上限"),
|
||||
({"max_total_s": 5}, "总时长上限"),
|
||||
({"enabled": "true"}, "开启数据剖析"),
|
||||
({"max_queries": True}, "查询次数上限"),
|
||||
({"table_max_queries": True}, "每张表最多查询数"),
|
||||
])
|
||||
def test_settings_problem_names_the_field_in_chinese(value, label):
|
||||
problem = profile_settings_problem({PROFILE_OPTION: value})
|
||||
@@ -51,9 +55,21 @@ def test_settings_problem_rejects_unknown_keys_and_wrong_shape():
|
||||
assert profile_settings_problem({PROFILE_OPTION: ["enabled"]})
|
||||
assert profile_settings_problem({}) is None
|
||||
assert profile_settings_problem({PROFILE_OPTION: None}) is None
|
||||
assert profile_settings_problem({PROFILE_OPTION: {"enabled": True, "max_queries": 500, "query_timeout_s": 1.5,
|
||||
"sample_size": 10, "max_scan_rows": 0,
|
||||
"max_total_s": 600}}) is None
|
||||
assert profile_settings_problem({PROFILE_OPTION: {"enabled": True, "table_max_queries": 200, "table_max_s": 10.5,
|
||||
"query_timeout_s": 1.5, "sample_size": 10,
|
||||
"max_scan_rows": 0}}) is None
|
||||
|
||||
|
||||
def test_legacy_whole_run_budgets_are_stripped_not_rejected():
|
||||
"""以前按整次剖析算的两项（查询次数上限、总时长上限）：保存时只摘掉，不报错；不折算成每张表的预算。"""
|
||||
old = {"charset": "utf8mb4", PROFILE_OPTION: {"enabled": True, "max_queries": 60, "max_total_s": 120,
|
||||
"sample_size": 500}}
|
||||
cleaned = catalog_profile.without_legacy(old)
|
||||
assert cleaned == {"charset": "utf8mb4", PROFILE_OPTION: {"enabled": True, "sample_size": 500}}
|
||||
assert old[PROFILE_OPTION]["max_queries"] == 60, "不改调用方的对象"
|
||||
assert profile_settings_problem(cleaned) is None
|
||||
plain = {PROFILE_OPTION: {"enabled": True}}
|
||||
assert catalog_profile.without_legacy(plain) is plain and catalog_profile.without_legacy(None) is None
|
||||
|
||||
|
||||
def test_profile_option_never_reaches_the_connection_url():
|
||||
|
||||
@@ -24,7 +24,7 @@ from app.db.models import DataSource
|
||||
from tests.fixtures.catalog import scenic, scenic_notes
|
||||
from tests.fixtures.catalog.fake_model import FakeModel
|
||||
from tests.test_catalog_patch import _schema
|
||||
from tests.test_catalog_profile import _notes, _on, _relation, _table, _url, client, make_source # noqa: F401
|
||||
from tests.test_catalog_profile import _notes, _on, _profile, _relation, _table, client, make_source # noqa: F401
|
||||
|
||||
AT = "2026-10-04T08:00:00+00:00"
|
||||
DAY1, DAY2, DAY3 = "2026-10-01T08:00:00+00:00", "2026-10-02T08:00:00+00:00", "2026-10-03T08:00:00+00:00"
|
||||
@@ -131,8 +131,7 @@ async def test_confirming_a_profiled_relation_by_patch_keeps_the_fanout_check(cl
|
||||
"""审查探针 A1：剖析得出多对一、覆盖率 1.0，经对话提案确认这条关系之后，扇出检查照样报 error。"""
|
||||
sql = "SELECT COUNT(g.id) AS n FROM gates g JOIN visits v ON v.gate_id = g.id"
|
||||
sid = await make_source(scenic_db, options=_on(), draft=["visits", "gates"])
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]})
|
||||
assert r.status_code == 200, r.text
|
||||
await _profile(client, sid, ["visits"])
|
||||
detail = await _notes(client, sid, "visits")
|
||||
rel = _relation(detail["notes"], "gate_id")
|
||||
assert (rel["source"], rel["status"], rel["cardinality"], rel["coverage"]) == (
|
||||
@@ -156,8 +155,7 @@ async def test_profiled_relation_without_child_check_only_warns(client, make_sou
|
||||
核实过：扇出检查只给 warning（「可能重复计算」），不给 error、不让出具降档。"""
|
||||
sql = "SELECT COUNT(g.id) AS n FROM gates g JOIN visits v ON v.gate_id = g.id"
|
||||
sid = await make_source(scenic_db, options=_on(max_scan_rows=100), draft=["visits", "gates"])
|
||||
r = await client.post(_url(sid), json={"tables": ["visits"]})
|
||||
assert r.status_code == 200, r.text
|
||||
await _profile(client, sid, ["visits"])
|
||||
rel = _relation((await _notes(client, sid, "visits"))["notes"], "gate_id")
|
||||
assert (rel["source"], rel["status"], rel["cardinality"]) == ("profile", "verified", "many_to_one")
|
||||
assert not rel.get("cardinality_checked")
|
||||
@@ -359,15 +357,13 @@ def test_codes_complete_flag_is_validated_and_counts_as_an_edit():
|
||||
|
||||
async def test_profile_marks_codes_complete_only_when_the_whole_table_was_read(client, make_source, scenic_db):
|
||||
whole = await make_source(scenic_db, options=_on(), draft=["visits"])
|
||||
r = await client.post(_url(whole), json={"tables": ["visits"]})
|
||||
assert r.status_code == 200, r.text
|
||||
await _profile(client, whole, ["visits"])
|
||||
codes = (await _notes(client, whole, "visits"))["notes"]["columns"]["status"]["codes"]
|
||||
assert "统计全表" in codes["note"] and codes.get("complete") is True
|
||||
|
||||
# 不做整表统计（只看前若干行）：取值可能没看全，不标
|
||||
capped = await make_source(scenic_db, options=_on(max_scan_rows=0, sample_size=10), draft=["visits"])
|
||||
r = await client.post(_url(capped), json={"tables": ["visits"]})
|
||||
assert r.status_code == 200, r.text
|
||||
await _profile(client, capped, ["visits"])
|
||||
codes = (await _notes(client, capped, "visits"))["notes"]["columns"]["status"]["codes"]
|
||||
assert "按前 10 行统计" in codes["note"] and "complete" not in codes
|
||||
|
||||
@@ -388,10 +384,10 @@ async def test_rerun_on_another_day_with_same_data_keeps_the_version(client, mak
|
||||
assert r.status_code == 200, r.text
|
||||
|
||||
monkeypatch.setattr(catalog, "now_iso", lambda: DAY1)
|
||||
first = _table((await client.post(_url(sid), json={"tables": ["visits"]})).json(), "visits")
|
||||
first = _table(await _profile(client, sid, ["visits"]), "visits")
|
||||
notes_day1 = (await _notes(client, sid, "visits"))["notes"]
|
||||
monkeypatch.setattr(catalog, "now_iso", lambda: DAY2)
|
||||
second = _table((await client.post(_url(sid), json={"tables": ["visits"]})).json(), "visits")
|
||||
second = _table(await _profile(client, sid, ["visits"]), "visits")
|
||||
assert (second["added"], second["updated"], second["removed"]) == (0, 0, 0)
|
||||
assert second["version"] == first["version"]
|
||||
assert (await _notes(client, sid, "visits"))["notes"] == notes_day1 # 备注里还是第一天的日期
|
||||
@@ -404,7 +400,7 @@ async def test_rerun_on_another_day_with_same_data_keeps_the_version(client, mak
|
||||
finally:
|
||||
db.close()
|
||||
monkeypatch.setattr(catalog, "now_iso", lambda: DAY3)
|
||||
third = _table((await client.post(_url(sid), json={"tables": ["visits"]})).json(), "visits")
|
||||
third = _table(await _profile(client, sid, ["visits"]), "visits")
|
||||
assert third["version"] == second["version"] + 1 and third["updated"] >= 1
|
||||
codes = (await _notes(client, sid, "visits"))["notes"]["columns"]["status"]["codes"]
|
||||
assert set(codes["value"]) == {"0", "1", "2"} and codes["updated_at"] == DAY3
|
||||
|
||||
Reference in New Issue
Block a user