mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 01:11:59 +00:00
数据目录起草改成后台任务：新表 catalog_draft_jobs、catalog_draft_tasks 存任务和逐表状态，同一个源同时只有一个排队或进行中的任务（接口先查、部分唯一索引兜底）；POST /catalog/draft-jobs 建任务（202），GET /draft-jobs/active、/latest、/{id} 读任务，/cancel 停止（排队的直接停，进行中的停掉正在进行的模型调用，已写进目录的不回滚），/retry 只重试失败、只写出一部分、停止了的表；同源 409 catalog_draft_busy、起草模型用不了 409 catalog_draft_model_unavailable、已结束 409 catalog_draft_finished；每张表一结束就并入目录，进度、计数、用量随时落库；服务启动时把没做完的任务标成中断、不自动续跑，关停时同样；目录清单每行加 last_draft，顶层加 guidance、active_job；数据源级的起草要求存在 options 里，经 PUT /catalog/guidance 维护（最多 2000 字，表名恰好叫 guidance 的整份提交照旧），数据源接口不显示、整份回写 options 时留着，拼连接串时摘掉；本次说明最多 1000 字，任务记下起草要求和本次说明的快照，同步起草也带上数据源的起草要求
This commit is contained in:
ThearchyHelios
+197
-14
@@ -7,11 +7,15 @@
|
||||
- POST /catalog/{table}/patch/preview 目录修改提案的预览（只读）：对着当前目录算改前、改后
|
||||
- POST /catalog/{table}/patch 保存目录修改提案（乐观锁），记为人工确认
|
||||
- GET /catalog/{table}/impact 影响面：引用这张表的已发布、受管模板（data/catalog_impact.py）
|
||||
- POST /catalog/draft 同步起草（注释、外键、命名推断，可选模型）
|
||||
- POST /catalog/draft 同步起草（注释、外键、命名推断，可选模型）。界面改用下面的后台起草任务，这个留着
|
||||
- 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）
|
||||
|
||||
规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径
|
||||
不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。
|
||||
起草任务的 409 带机读码（api/coded.py）：同一个源已有排队或进行中的任务、起草模型用不了、要停止的任务已经结束。
|
||||
|
||||
上传表格的源按当前快照的表结构回答（table_versions.resolve_source），和查询实际用的库是同一个版本。
|
||||
"""
|
||||
@@ -19,13 +23,17 @@ from __future__ import annotations
|
||||
|
||||
from typing import Any, Literal
|
||||
|
||||
from fastapi import APIRouter, Depends, Header, HTTPException
|
||||
from pydantic import BaseModel, Field
|
||||
from fastapi import APIRouter, Body, Depends, Header, HTTPException
|
||||
from fastapi.exceptions import RequestValidationError
|
||||
from pydantic import BaseModel, Field, ValidationError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.api.coded import PROFILE_BUSY, SCHEMA_MISSING, SNAPSHOT_TAMPERED, CodedHTTPException
|
||||
from app.api.coded import (
|
||||
CATALOG_DRAFT_BUSY, CATALOG_DRAFT_FINISHED, CATALOG_DRAFT_MODEL_UNAVAILABLE, PROFILE_BUSY, SCHEMA_MISSING,
|
||||
SNAPSHOT_TAMPERED, CodedHTTPException,
|
||||
)
|
||||
from app.api.runs import actor_of
|
||||
from app.data import catalog, catalog_impact, catalog_profile, introspect, table_versions
|
||||
from app.data import catalog, catalog_impact, catalog_jobs, catalog_profile, introspect, table_versions
|
||||
from app.data.engine import SnapshotTampered
|
||||
from app.db.base import get_session
|
||||
from app.db.models import DataSource
|
||||
@@ -64,10 +72,25 @@ class CatalogPatchIn(CatalogPatchPreviewIn):
|
||||
class CatalogDraftIn(BaseModel):
|
||||
#: 要起草的表；不给或为空时按使用次数取前 20 张
|
||||
tables: list[str] | None = Field(default=None, max_length=200)
|
||||
#: 是否请助手的模型起草中文名、粒度、列的含义等
|
||||
#: 是否请起草模型（设置里的起草模型，没设沿用助手的模型）写中文名、粒度、列的含义等
|
||||
use_model: bool = False
|
||||
|
||||
|
||||
class DraftJobIn(BaseModel):
|
||||
#: 要起草的表；不给或为空时按使用次数取前 20 张。最多 500 张（超了 422，说明是中文）
|
||||
tables: list[str] | None = None
|
||||
#: 是否请起草模型写中文名、粒度、列的含义等
|
||||
use_model: bool = False
|
||||
#: 这一次的补充说明，和数据源级的起草要求一起交给模型（最多 1000 字）
|
||||
note: str | None = None
|
||||
|
||||
|
||||
class GuidanceIn(BaseModel):
|
||||
#: 数据源级的起草要求；空串（或 null）清除。最多 2000 字（超了 422，说明是中文）。必须写这一项：表名恰好叫
|
||||
#: guidance 时，整份提交那张表目录的请求也落到这个路径上，不能把它当成「清除起草要求」
|
||||
guidance: str | None
|
||||
|
||||
|
||||
class CatalogProfileIn(BaseModel):
|
||||
#: 要剖析的表；不给或为空时按使用次数取前 10 张目录里有待核实关系的表
|
||||
tables: list[str] | None = Field(default=None, max_length=200)
|
||||
@@ -100,7 +123,7 @@ 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) -> dict[str, Any]:
|
||||
usage: int, last_draft: 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")
|
||||
@@ -120,6 +143,8 @@ def _summary_row(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogE
|
||||
"version": entry.version if entry else 0,
|
||||
"updated_at": _iso(entry.updated_at) if entry else None,
|
||||
"updated_by": entry.updated_by if entry else None,
|
||||
# 这张表最近一次起草任务里的那一项：{status, job_id, finished_at, error, model_error}，没起草过是 null
|
||||
"last_draft": last_draft,
|
||||
}
|
||||
|
||||
|
||||
@@ -180,15 +205,18 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi
|
||||
说明有没有截断，schema_total 是数据库里一共几张表（含视图），界面据此说明「只探查了前 200 张」。老缓存没记
|
||||
总数时按探查到的张数算。
|
||||
"""
|
||||
_, source = await _resolved(session, source_id)
|
||||
row, source = await _resolved(session, source_id)
|
||||
cache = source.schema_cache or {}
|
||||
tables = cache.get("tables") or {}
|
||||
entries = await catalog.read_catalog(session, source.id)
|
||||
usage = await catalog.table_usage(session, source)
|
||||
drafts = await catalog_jobs.last_drafts(session, source.id)
|
||||
order = {name: i for i, name in enumerate(tables)}
|
||||
rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0)) for name, meta in tables.items()]
|
||||
rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0), drafts.get(name))
|
||||
for name, meta in tables.items()]
|
||||
rows.sort(key=lambda r: (-r["usage"], order[r["table_name"]]))
|
||||
rows += [_summary_row(name, None, entry, 0) for name, entry in sorted(entries.items()) if name not in tables]
|
||||
rows += [_summary_row(name, None, entry, 0, drafts.get(name))
|
||||
for name, entry in sorted(entries.items()) if name not in tables]
|
||||
return {
|
||||
"tables": rows,
|
||||
"system_notes": catalog.system_notes_source(source),
|
||||
@@ -196,6 +224,10 @@ async def list_catalog(source_id: str, session: AsyncSession = Depends(get_sessi
|
||||
"schema_note": None if tables else introspect.why_empty(cache),
|
||||
"schema_truncated": bool(cache.get("truncated")),
|
||||
"schema_total": max(_int(cache.get("total")), len(tables)),
|
||||
# 数据源级的起草要求（没有是 null）
|
||||
"guidance": catalog_jobs.guidance_of(row),
|
||||
# 排队或进行中的起草任务 {id, status, total, done}，界面据此显示进度；没有是 null
|
||||
"active_job": await catalog_jobs.job_brief(session, await catalog_jobs.active_job(session, source.id)),
|
||||
}
|
||||
|
||||
|
||||
@@ -207,6 +239,155 @@ async def get_catalog_table(source_id: str, table: str,
|
||||
return await _detail(session, source, key, entry)
|
||||
|
||||
|
||||
async def _source_row(session: AsyncSession, source_id: str) -> DataSource:
|
||||
row = await session.get(DataSource, source_id)
|
||||
if row is None:
|
||||
raise HTTPException(404, "数据源不存在，可能已被删除")
|
||||
return row
|
||||
|
||||
|
||||
@router.put("/{source_id}/catalog/guidance")
|
||||
async def put_catalog_guidance(source_id: str, body: dict[str, Any] = Body(...),
|
||||
x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> Any:
|
||||
"""存数据源级的起草要求：去首尾空白，空串清除，最多 2000 字（超了 422）。上传表格的源存在它自己的数据源记录上。
|
||||
|
||||
要登记在 PUT /catalog/{table} 前面：不然「guidance」会被当成表名。反过来，表名恰好叫 guidance 时，整份提交那张表
|
||||
目录的请求（{notes, if_version}，没有 guidance 这一项）也落到这里：照旧当作那张表的整份提交。"""
|
||||
if "guidance" not in body and "notes" in body:
|
||||
try:
|
||||
table_payload = CatalogPutIn.model_validate(body)
|
||||
except ValidationError as e:
|
||||
raise RequestValidationError(e.errors()) from e
|
||||
return await put_catalog_table(source_id, "guidance", table_payload, x_actor=x_actor, session=session)
|
||||
try:
|
||||
payload = GuidanceIn.model_validate(body)
|
||||
except ValidationError as e:
|
||||
raise RequestValidationError(e.errors()) from e
|
||||
row = await _source_row(session, source_id)
|
||||
try:
|
||||
value = await catalog_jobs.save_guidance(session, row, payload.guidance)
|
||||
except catalog_jobs.TextTooLong as e:
|
||||
raise HTTPException(422, str(e)) from e
|
||||
return {"guidance": value}
|
||||
|
||||
|
||||
async def _ensure_model(session: AsyncSession) -> None:
|
||||
"""起草模型用不了（没有接入、停用、只有演示接入……）：409，不建任务。说法同同步起草时的 model_error。"""
|
||||
try:
|
||||
await catalog_jobs.check_model(session)
|
||||
except catalog.CatalogModelUnavailable as e:
|
||||
raise CodedHTTPException(409, str(e), CATALOG_DRAFT_MODEL_UNAVAILABLE) from e
|
||||
|
||||
|
||||
def _no_schema(cache: dict[str, Any]) -> HTTPException:
|
||||
return HTTPException(409, f"{introspect.why_empty(cache)}，无法起草数据目录。请先在数据源卡片上点「探查结构」")
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/draft-jobs", status_code=202)
|
||||
async def create_draft_job(source_id: str, payload: DraftJobIn, x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""建后台起草任务，马上返回（202，任务是排队中）。同一个源已有排队或进行中的任务 409 catalog_draft_busy（界面改去
|
||||
GET /draft-jobs/active 显示那一个）；用模型而起草模型用不了 409 catalog_draft_model_unavailable；没有表结构 409。
|
||||
表结构里没有的表照样建进任务、状态 skipped。"""
|
||||
if payload.tables and len(payload.tables) > catalog_jobs.MAX_TABLES:
|
||||
raise HTTPException(422, f"一次最多起草 {catalog_jobs.MAX_TABLES} 张表，当前 {len(payload.tables)} 张，"
|
||||
"请分几次起草")
|
||||
try:
|
||||
note = catalog_jobs.clean_text(payload.note, catalog_jobs.NOTE_MAX, "本次说明")
|
||||
except catalog_jobs.TextTooLong as e:
|
||||
raise HTTPException(422, str(e)) from e
|
||||
row, source = await _resolved(session, source_id)
|
||||
try:
|
||||
await catalog_jobs.ensure_idle(session, row)
|
||||
except catalog_jobs.DraftBusy as e:
|
||||
raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
|
||||
cache = source.schema_cache or {}
|
||||
if not cache.get("tables"):
|
||||
raise _no_schema(cache)
|
||||
if payload.use_model:
|
||||
await _ensure_model(session)
|
||||
try:
|
||||
job, tasks = await catalog_jobs.create_job(session, row, source, tables=payload.tables or None,
|
||||
use_model=payload.use_model, note=note, actor=actor_of(x_actor))
|
||||
except catalog_jobs.DraftBusy as e:
|
||||
raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
|
||||
catalog_jobs.manager.start(job.id)
|
||||
return {"job": catalog_jobs.job_dict(job, tasks)}
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/draft-jobs/active")
|
||||
async def active_draft_job(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""排队或进行中的那个起草任务，没有是 null。"""
|
||||
row = await _source_row(session, source_id)
|
||||
return {"job": await catalog_jobs.job_out(session, await catalog_jobs.active_job(session, row.id))}
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/draft-jobs/latest")
|
||||
async def latest_draft_job(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""最近建的一个起草任务（任何状态），没有是 null。"""
|
||||
row = await _source_row(session, source_id)
|
||||
return {"job": await catalog_jobs.job_out(session, await catalog_jobs.latest_job(session, row.id))}
|
||||
|
||||
|
||||
async def _job_or_404(session: AsyncSession, source_id: str, job_id: str) -> tuple[DataSource, Any]:
|
||||
row = await _source_row(session, source_id)
|
||||
job = await catalog_jobs.load(session, row.id, job_id)
|
||||
if job is None:
|
||||
raise HTTPException(404, "起草任务不存在，可能属于别的数据源，或者已随数据源删除")
|
||||
return row, job
|
||||
|
||||
|
||||
@router.get("/{source_id}/catalog/draft-jobs/{job_id}")
|
||||
async def get_draft_job(source_id: str, job_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
_, job = await _job_or_404(session, source_id, job_id)
|
||||
return {"job": await catalog_jobs.job_out(session, job)}
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/draft-jobs/{job_id}/cancel")
|
||||
async def cancel_draft_job(source_id: str, job_id: str,
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""停止：排队的直接停止；进行中的停掉正在进行的模型调用，没开始和没写完的表记停止，已经写进目录的不回滚。
|
||||
已经结束的 409 catalog_draft_finished。返回停止之后的任务。"""
|
||||
row, job = await _job_or_404(session, source_id, job_id)
|
||||
owner = row.id
|
||||
try:
|
||||
await catalog_jobs.cancel(session, job)
|
||||
except catalog_jobs.DraftFinished as e:
|
||||
raise CodedHTTPException(409, str(e), CATALOG_DRAFT_FINISHED) from e
|
||||
session.expire_all() # 后台在别的会话里收的尾：重新读
|
||||
return {"job": await catalog_jobs.job_out(session, await catalog_jobs.load(session, owner, job_id))}
|
||||
|
||||
|
||||
@router.post("/{source_id}/catalog/draft-jobs/{job_id}/retry", status_code=202)
|
||||
async def retry_draft_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）的表；用不用模型、本次说明沿用，起草要求
|
||||
取数据源当前的。有排队或进行中的任务 409 catalog_draft_busy；没有可重试的表 422；起草模型用不了 409。"""
|
||||
_, old = await _job_or_404(session, source_id, job_id)
|
||||
row, source = await _resolved(session, source_id)
|
||||
try:
|
||||
await catalog_jobs.ensure_idle(session, row)
|
||||
except catalog_jobs.DraftBusy as e:
|
||||
raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
|
||||
try:
|
||||
tables = await catalog_jobs.retry_tables(session, old)
|
||||
except catalog_jobs.NothingToRetry as e:
|
||||
raise HTTPException(422, str(e)) from e
|
||||
cache = source.schema_cache or {}
|
||||
if not cache.get("tables"):
|
||||
raise _no_schema(cache)
|
||||
if old.use_model:
|
||||
await _ensure_model(session)
|
||||
try:
|
||||
job, tasks = await catalog_jobs.create_job(session, row, source, tables=tables, use_model=old.use_model,
|
||||
note=old.note, actor=actor_of(x_actor), retry_of=old.id)
|
||||
except catalog_jobs.DraftBusy as e:
|
||||
raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
|
||||
catalog_jobs.manager.start(job.id)
|
||||
return {"job": catalog_jobs.job_dict(job, tasks)}
|
||||
|
||||
|
||||
@router.put("/{source_id}/catalog/{table}")
|
||||
async def put_catalog_table(source_id: str, table: str, payload: CatalogPutIn,
|
||||
x_actor: str | None = Header(default=None),
|
||||
@@ -228,17 +409,19 @@ async def put_catalog_table(source_id: str, table: str, payload: CatalogPutIn,
|
||||
async def draft_catalog(source_id: str, payload: CatalogDraftIn,
|
||||
x_actor: str | None = Header(default=None),
|
||||
session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
|
||||
"""同步起草。返回每张表新增、更新、删除了几项；模型用不了或某张表的模型起草失败，照实写在结果里。
|
||||
"""同步起草（界面改用后台起草任务，这个接口留着，返回形状不变）。引擎同后台任务（分段、按失败原因处理），起草要求
|
||||
取数据源级的那份。返回每张表新增、更新、删除了几项；模型用不了或某张表的模型起草失败，照实写在结果里。
|
||||
|
||||
model_items：模型给了这张表几项可用内容（没用模型、模型这部分失败时为 null）。界面据此区分「模型给了内容、
|
||||
只是和现有目录一致」（model_items > 0、三个数都是 0）和「模型没给内容」（model_error 有值）。
|
||||
"""
|
||||
_, source = await _resolved(session, source_id)
|
||||
row, source = await _resolved(session, source_id)
|
||||
cache = source.schema_cache or {}
|
||||
if not cache.get("tables"):
|
||||
raise HTTPException(409, f"{introspect.why_empty(cache)}，无法起草数据目录。请先在数据源卡片上点「探查结构」")
|
||||
raise _no_schema(cache)
|
||||
report = await catalog.draft_catalog(session, source, tables=payload.tables or None,
|
||||
use_model=payload.use_model, actor=actor_of(x_actor))
|
||||
use_model=payload.use_model, actor=actor_of(x_actor),
|
||||
guidance=catalog_jobs.guidance_of(row))
|
||||
rows = [{"table_name": r.table, "added": r.added, "updated": r.updated, "removed": r.removed,
|
||||
"version": r.version, "error": r.error, "model_error": r.model_error, "model_items": r.model_items}
|
||||
for r in report.tables]
|
||||
|
||||
@@ -19,6 +19,11 @@ DATASOURCE_INACTIVE = "datasource_inactive"
|
||||
SCHEMA_MISSING = "schema_missing"
|
||||
PROFILE_BUSY = "profile_busy"
|
||||
SNAPSHOT_TAMPERED = "snapshot_tampered"
|
||||
#: 数据目录的后台起草任务（api/catalog.py 的 /catalog/draft-jobs）：同一个源已有排队或进行中的任务（409，界面改去
|
||||
#: GET /draft-jobs/active 显示那一个）、起草模型用不了（409，不建任务）、要停止的任务已经结束了（409）
|
||||
CATALOG_DRAFT_BUSY = "catalog_draft_busy"
|
||||
CATALOG_DRAFT_MODEL_UNAVAILABLE = "catalog_draft_model_unavailable"
|
||||
CATALOG_DRAFT_FINISHED = "catalog_draft_finished"
|
||||
|
||||
|
||||
class CodedHTTPException(HTTPException):
|
||||
|
||||
@@ -27,8 +27,8 @@ from app.data.catalog_profile import profile_settings_rejection
|
||||
from app.data import introspect as introspect_mod
|
||||
from app.data import table_versions
|
||||
from app.data.engine import (
|
||||
MASK_COLUMNS_OPTION, MAX_QUERY_TIMEOUT_S, QUERY_TIMEOUT_OPTION, SUPPORTED_KINDS, SnapshotTampered, build_url,
|
||||
engine_args, engines, mask_columns_problem, query_timeout_problem,
|
||||
CATALOG_DRAFT_GUIDANCE_OPTION, MASK_COLUMNS_OPTION, MAX_QUERY_TIMEOUT_S, QUERY_TIMEOUT_OPTION, SUPPORTED_KINDS,
|
||||
SnapshotTampered, build_url, engine_args, engines, mask_columns_problem, query_timeout_problem,
|
||||
)
|
||||
from app.db.base import get_session, new_id
|
||||
from app.db.models import DataSource, ImportStaging, SourceSnapshot, TableImport, TableRecipe
|
||||
@@ -124,6 +124,22 @@ def _import_mode(row: DataSource) -> str | None:
|
||||
return "simple" if _is_upload(row) else None
|
||||
|
||||
|
||||
#: options 里由别的接口维护的项：数据源表单不显示、不提交它们，整份回写 options 时原样留着（同模型接入 extra 里的
|
||||
#: last_check）。数据目录的起草要求只经 PUT /catalog/guidance 改（api/catalog.py）：表单把认不得的项放进「高级参数」
|
||||
#: 的单行输入框，多行的要求会被压成一行
|
||||
_MANAGED_OPTIONS = (CATALOG_DRAFT_GUIDANCE_OPTION,)
|
||||
|
||||
|
||||
def _shown_options(options: dict[str, Any] | None) -> dict[str, Any]:
|
||||
return {k: v for k, v in (options or {}).items() if k not in _MANAGED_OPTIONS}
|
||||
|
||||
|
||||
def _with_managed(options: dict[str, Any], saved: dict[str, Any] | None) -> dict[str, Any]:
|
||||
"""表单提交的 options + 存着的、由别的接口维护的项（提交里带了也不认）。"""
|
||||
kept = {k: v for k, v in (saved or {}).items() if k in _MANAGED_OPTIONS}
|
||||
return {**_shown_options(options), **kept}
|
||||
|
||||
|
||||
def _to_out(row: DataSource, snapshot: dict[str, Any] | None = None,
|
||||
recipe_info: dict[str, Any] | None = None) -> DataSourceOut:
|
||||
cache = row.schema_cache or {}
|
||||
@@ -137,7 +153,7 @@ def _to_out(row: DataSource, snapshot: dict[str, Any] | None = None,
|
||||
schema_error=str(cache.get("error") or "") if cache.get("failed") else "",
|
||||
available_schemas=list(cache.get("available_schemas") or []),
|
||||
id=row.id, name=row.name, kind=row.kind, host=row.host, port=row.port,
|
||||
database=row.database, username=row.username, options=row.options or {},
|
||||
database=row.database, username=row.username, options=_shown_options(row.options),
|
||||
readonly=row.readonly, description=row.description, enabled=row.enabled,
|
||||
password_masked=mask(row.password),
|
||||
has_password=bool(row.password),
|
||||
@@ -429,7 +445,8 @@ async def create_source(
|
||||
_refuse_reserved(payload.kind, payload.database)
|
||||
|
||||
row = DataSource(
|
||||
**payload.model_dump(exclude={"password"}),
|
||||
**payload.model_dump(exclude={"password", "options"}),
|
||||
options=_with_managed(payload.options, None),
|
||||
password=encrypt(payload.password),
|
||||
)
|
||||
_settle_oracle(row, None, None)
|
||||
@@ -472,6 +489,7 @@ async def update_source(
|
||||
raise HTTPException(422, _UPLOAD_DIR_READONLY)
|
||||
if "options" in data:
|
||||
_refuse_bad_options(data["options"])
|
||||
data["options"] = _with_managed(data["options"] or {}, row.options)
|
||||
if "password" in data:
|
||||
# 空串表示清空，非空表示换新的
|
||||
data["password"] = encrypt(data["password"]) if data["password"] else None
|
||||
|
||||
@@ -0,0 +1,639 @@
|
||||
"""数据目录的后台起草任务，以及数据源级的起草要求。
|
||||
|
||||
以前起草是一次同步长请求（POST /catalog/draft，界面每 12 张表一个请求）：一个请求出错整个对话框就停，关掉对话框
|
||||
之后没有任何记录哪些表失败、为什么。现在：
|
||||
|
||||
- **建任务**（create_job）：同一个数据源同时只有一个排队或进行中的任务（DraftBusy；部分唯一索引兜底）。不指定表时
|
||||
按使用次数取前 20 张；表结构里没有的表照样建进任务、状态 skipped。起草要求（数据源级）和本次说明随任务记下快照。
|
||||
- **在后台起草**（manager.start）：进程内的 asyncio 任务，引擎是 data/catalog_draft.py（分段、按失败原因处理）。
|
||||
每张表一结束就并入目录（catalog.write_draft，规则和同步起草一样：人工确认、驳回过的不动，乐观锁冲突重读重并）；
|
||||
逐表的状态、到第几段、计数、用量随时落库，接口读的是库里的状态。
|
||||
- **停止**（cancel）：排队的直接 cancelled；进行中的停掉正在进行的模型调用，没开始和没写完的表记 cancelled，已经写进
|
||||
目录的不回滚，任务最后是 cancelled。
|
||||
- **重试**（retry_tables + create_job）：新任务，只含原任务里 failed、partial、cancelled 的表；用不用模型、本次说明沿用，
|
||||
起草要求取数据源当前的。
|
||||
- **服务重启**：启动时（startup）把排队、进行中的任务标成 interrupted，没结束的表记 cancelled、写明「服务重启，起草
|
||||
中断」。不自动续跑：复制出来的数据目录（检查、验收）启动时不能自己去调模型；界面让人点「重试」续上。关停时
|
||||
（manager.shutdown）停掉进行中的任务，同样标成 interrupted。
|
||||
|
||||
任务的库写入一律经 _JobRun._write：一条一条来，shield 起来——停止任务时不会打断写到一半的库操作，收尾前等它们写完。
|
||||
|
||||
**起草要求**存在数据源 options 的 catalog_draft_guidance 里（上传表格的源存在它自己的 DataSource 行上），一段文字，
|
||||
去首尾空白、最多 GUIDANCE_MAX 字，空串等于没有。只经 PUT /catalog/guidance 改：数据源接口不显示它、整份回写 options
|
||||
时留着（api/datasources.py），拼连接串时摘掉（data/engine.py）。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import func, select, update
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.data import catalog, catalog_draft
|
||||
from app.data.engine import CATALOG_DRAFT_GUIDANCE_OPTION
|
||||
from app.db.base import SessionLocal, new_id, utcnow
|
||||
from app.db.models import CatalogDraftJob, CatalogDraftTask, DataSource
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
#: 任务还没结束的状态（同一个源只能有一个）、结束了的状态
|
||||
ACTIVE = ("queued", "running")
|
||||
FINISHED = ("succeeded", "partial", "failed", "cancelled", "interrupted")
|
||||
#: 逐表的状态，以及其中算「已结束」的、重试时要带上的
|
||||
TASK_STATUSES = ("pending", "running", "ok", "partial", "failed", "skipped", "cancelled")
|
||||
TASK_DONE = ("ok", "partial", "failed", "skipped", "cancelled")
|
||||
RETRYABLE = ("failed", "partial", "cancelled")
|
||||
_OPEN = ("pending", "running")
|
||||
|
||||
#: 一个任务最多几张表；起草要求、本次说明最多几个字
|
||||
MAX_TABLES = 500
|
||||
GUIDANCE_MAX = 2000
|
||||
NOTE_MAX = 1000
|
||||
#: 目录清单上「这张表最近一次起草」只看最近这么多个任务
|
||||
LAST_DRAFT_JOBS = 100
|
||||
#: 停止进行中的任务时最多等它收尾多久（秒）；关停服务时另给（shutdown 的 grace）
|
||||
STOP_GRACE_S = 10.0
|
||||
|
||||
INTERRUPTED_ERROR = "服务重启，起草中断"
|
||||
UNFINISHED_ERROR = "起草结束时这张表还没有写完，请重试这张表"
|
||||
WRITE_ERROR = "写入数据目录时出错：{line}"
|
||||
JOB_ERROR = "起草任务出错：{line}"
|
||||
SOURCE_GONE = "数据源不存在，可能已被删除"
|
||||
|
||||
|
||||
class DraftBusy(Exception):
|
||||
"""这个数据源已经有排队或进行中的起草任务。message 给人看。"""
|
||||
|
||||
|
||||
class DraftFinished(Exception):
|
||||
"""要停止的任务已经结束了。"""
|
||||
|
||||
|
||||
class NothingToRetry(Exception):
|
||||
"""原任务里没有需要重试的表。"""
|
||||
|
||||
|
||||
class TextTooLong(ValueError):
|
||||
"""起草要求、本次说明超长。message 给人看。"""
|
||||
|
||||
|
||||
class _JobFailed(Exception):
|
||||
"""任务整体出错（数据源没了、快照没了、模型用不了）：message 就是任务的 error。"""
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 起草要求、本次说明
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
def clean_text(value: Any, limit: int, label: str) -> str | None:
|
||||
"""去首尾空白；空串（和不是字符串的）等于没有；超过 limit 个字抛 TextTooLong。"""
|
||||
text = value.strip() if isinstance(value, str) else ""
|
||||
if len(text) > limit:
|
||||
raise TextTooLong(f"{label}最多 {limit} 字，当前 {len(text)} 字，请删减后再保存")
|
||||
return text or None
|
||||
|
||||
|
||||
def guidance_of(row: Any) -> str | None:
|
||||
"""数据源级的起草要求（没有是 None）。"""
|
||||
raw = (getattr(row, "options", None) or {}).get(CATALOG_DRAFT_GUIDANCE_OPTION)
|
||||
return raw.strip() or None if isinstance(raw, str) else None
|
||||
|
||||
|
||||
async def save_guidance(session: AsyncSession, row: DataSource, value: Any) -> str | None:
|
||||
"""存起草要求，返回存下的（清除了是 None）。超长抛 TextTooLong，什么都不改。"""
|
||||
text = clean_text(value, GUIDANCE_MAX, "起草要求")
|
||||
options = {k: v for k, v in (row.options or {}).items() if k != CATALOG_DRAFT_GUIDANCE_OPTION}
|
||||
if text:
|
||||
options[CATALOG_DRAFT_GUIDANCE_OPTION] = text
|
||||
row.options = options
|
||||
await session.commit()
|
||||
return text
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 读任务
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
def _iso(value: Any) -> str | None:
|
||||
return value.isoformat() if value is not None else None
|
||||
|
||||
|
||||
async def active_job(session: AsyncSession, source_id: str) -> CatalogDraftJob | None:
|
||||
"""排队或进行中的那个（同一个源最多一个）。"""
|
||||
return (await session.execute(
|
||||
select(CatalogDraftJob).where(CatalogDraftJob.source_id == source_id, CatalogDraftJob.status.in_(ACTIVE))
|
||||
.order_by(CatalogDraftJob.created_at.desc()).limit(1)
|
||||
)).scalar_one_or_none()
|
||||
|
||||
|
||||
async def latest_job(session: AsyncSession, source_id: str) -> CatalogDraftJob | None:
|
||||
"""最近建的一个（任何状态）。"""
|
||||
return (await session.execute(
|
||||
select(CatalogDraftJob).where(CatalogDraftJob.source_id == source_id)
|
||||
.order_by(CatalogDraftJob.created_at.desc()).limit(1)
|
||||
)).scalar_one_or_none()
|
||||
|
||||
|
||||
async def load(session: AsyncSession, source_id: str, job_id: str) -> CatalogDraftJob | None:
|
||||
"""这个源的这个任务；不存在、不属于这个源都是 None。"""
|
||||
job = await session.get(CatalogDraftJob, job_id)
|
||||
return job if job is not None and job.source_id == source_id else None
|
||||
|
||||
|
||||
async def tasks_of(session: AsyncSession, job_id: str) -> list[CatalogDraftTask]:
|
||||
return list((await session.execute(
|
||||
select(CatalogDraftTask).where(CatalogDraftTask.job_id == job_id).order_by(CatalogDraftTask.seq)
|
||||
)).scalars())
|
||||
|
||||
|
||||
def task_dict(task: CatalogDraftTask) -> dict[str, Any]:
|
||||
return {
|
||||
"table_name": task.table_name,
|
||||
"status": task.status,
|
||||
"columns": task.column_count,
|
||||
"parts_total": task.parts_total,
|
||||
"parts_done": task.parts_done,
|
||||
"added": task.added,
|
||||
"updated": task.updated,
|
||||
"removed": task.removed,
|
||||
"version": task.version,
|
||||
"model_items": task.model_items,
|
||||
"error": task.error,
|
||||
"model_error": task.model_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: CatalogDraftJob, tasks: list[CatalogDraftTask]) -> dict[str, Any]:
|
||||
"""接口返回的任务（CatalogDraftJob）：逐表的状态按起草顺序；计数、合计都按逐表的算。"""
|
||||
counts = _counts([t.status for t in tasks])
|
||||
return {
|
||||
"id": job.id,
|
||||
"source_id": job.source_id,
|
||||
"status": job.status,
|
||||
"use_model": bool(job.use_model),
|
||||
"model": job.model,
|
||||
"guidance": job.guidance,
|
||||
"note": job.note,
|
||||
"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,
|
||||
"total": len(tasks),
|
||||
"done": sum(counts.get(s, 0) for s in TASK_DONE),
|
||||
"counts": counts,
|
||||
"added": sum(t.added or 0 for t in tasks),
|
||||
"updated": sum(t.updated or 0 for t in tasks),
|
||||
"removed": sum(t.removed or 0 for t in tasks),
|
||||
"usage": {"calls": job.calls or 0, "input_tokens": job.input_tokens or 0,
|
||||
"output_tokens": job.output_tokens or 0},
|
||||
"error": job.error,
|
||||
"tasks": [task_dict(t) for t in tasks],
|
||||
}
|
||||
|
||||
|
||||
async def job_out(session: AsyncSession, job: CatalogDraftJob | None) -> dict[str, Any] | None:
|
||||
return None if job is None else job_dict(job, await tasks_of(session, job.id))
|
||||
|
||||
|
||||
async def job_brief(session: AsyncSession, job: CatalogDraftJob | None) -> dict[str, Any] | None:
|
||||
"""目录清单顶上的进度：{id, status, total, done}。"""
|
||||
if job is None:
|
||||
return None
|
||||
rows = (await session.execute(
|
||||
select(CatalogDraftTask.status, func.count()).where(CatalogDraftTask.job_id == job.id)
|
||||
.group_by(CatalogDraftTask.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)}
|
||||
|
||||
|
||||
async def last_drafts(session: AsyncSession, source_id: str) -> dict[str, dict[str, Any]]:
|
||||
"""每张表最近一次起草任务里的那一项：{表名: {status, job_id, finished_at, error, model_error}}。任务还在进行，
|
||||
就是进行中的状态。只看最近 LAST_DRAFT_JOBS 个任务。"""
|
||||
ids = list((await session.execute(
|
||||
select(CatalogDraftJob.id).where(CatalogDraftJob.source_id == source_id)
|
||||
.order_by(CatalogDraftJob.created_at.desc()).limit(LAST_DRAFT_JOBS)
|
||||
)).scalars())
|
||||
if not ids:
|
||||
return {}
|
||||
rank = {job_id: i for i, job_id in enumerate(ids)}
|
||||
rows = (await session.execute(select(CatalogDraftTask).where(CatalogDraftTask.job_id.in_(ids)))).scalars()
|
||||
out: dict[str, dict[str, Any]] = {}
|
||||
for task in sorted(rows, key=lambda t: (rank[t.job_id], t.seq)):
|
||||
out.setdefault(task.table_name, {"status": task.status, "job_id": task.job_id,
|
||||
"finished_at": _iso(task.finished_at), "error": task.error,
|
||||
"model_error": task.model_error})
|
||||
return out
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 建任务、重试、停止
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
def busy_message(row: Any) -> str:
|
||||
return f"数据源「{getattr(row, 'name', '')}」正在起草数据目录，请等这次起草结束（或先停止它）再开始新的起草"
|
||||
|
||||
|
||||
async def ensure_idle(session: AsyncSession, row: Any) -> None:
|
||||
"""这个源没有排队或进行中的任务；有就抛 DraftBusy。"""
|
||||
if await active_job(session, row.id) is not None:
|
||||
raise DraftBusy(busy_message(row))
|
||||
|
||||
|
||||
async def check_model(session: AsyncSession) -> None:
|
||||
"""起草模型能不能用：不能用抛 catalog.CatalogModelUnavailable（只建模型对象，不发请求）。"""
|
||||
await catalog.resolve_draft_model(session)
|
||||
|
||||
|
||||
async def create_job(session: AsyncSession, row: Any, source: Any, *, tables: list[str] | None, use_model: bool,
|
||||
note: str | None, actor: str | None,
|
||||
retry_of: str | None = None) -> tuple[CatalogDraftJob, list[CatalogDraftTask]]:
|
||||
"""建一个排队的任务（还没开始，调用方接着 manager.start）。row 是数据源记录（起草要求取它的），source 是读表结构
|
||||
用的源（上传源是绑定当前快照的视图）。有排队或进行中的任务抛 DraftBusy。"""
|
||||
await ensure_idle(session, row)
|
||||
cache = getattr(source, "schema_cache", None) or {}
|
||||
all_tables = cache.get("tables") or {}
|
||||
wanted = list(tables) if tables else await catalog.default_tables(session, source)
|
||||
job = CatalogDraftJob(id=new_id(), source_id=row.id, status="queued", use_model=bool(use_model),
|
||||
guidance=guidance_of(row), note=note, created_by=actor, retry_of=retry_of)
|
||||
tasks: list[CatalogDraftTask] = []
|
||||
seen: set[str] = set()
|
||||
now = utcnow()
|
||||
for name in dict.fromkeys(str(t) for t in wanted):
|
||||
key = catalog.resolve_table_name(cache, name)
|
||||
if key is None:
|
||||
tasks.append(CatalogDraftTask(job_id=job.id, seq=len(tasks), table_name=name, status="skipped",
|
||||
error=catalog.missing_table_error(name), finished_at=now))
|
||||
continue
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
columns = len((all_tables.get(key) or {}).get("columns") or [])
|
||||
tasks.append(CatalogDraftTask(job_id=job.id, seq=len(tasks), table_name=key, status="pending",
|
||||
column_count=columns,
|
||||
parts_total=catalog_draft.parts_for(columns) if use_model else None))
|
||||
session.add(job)
|
||||
try:
|
||||
await session.flush()
|
||||
session.add_all(tasks)
|
||||
await session.commit()
|
||||
except IntegrityError as e:
|
||||
# 部分唯一索引：另一个请求同时给这个源建了任务
|
||||
await session.rollback()
|
||||
raise DraftBusy(busy_message(row)) from e
|
||||
return job, tasks
|
||||
|
||||
|
||||
async def retry_tables(session: AsyncSession, job: CatalogDraftJob) -> list[str]:
|
||||
"""重试要带上的表：原任务里 failed、partial、cancelled 的，按原来的顺序。没有抛 NothingToRetry。"""
|
||||
names = [t.table_name for t in await tasks_of(session, job.id) if t.status in RETRYABLE]
|
||||
if not names:
|
||||
raise NothingToRetry("这次起草里没有失败、只写出一部分或停止了的表，不需要重试")
|
||||
return names
|
||||
|
||||
|
||||
async def cancel(session: AsyncSession, job: CatalogDraftJob) -> None:
|
||||
"""停止一个任务。排队的直接 cancelled；进行中的交给 manager 停（停掉模型调用、没写完的表记 cancelled）。
|
||||
已经结束的抛 DraftFinished。"""
|
||||
if job.status in FINISHED:
|
||||
raise DraftFinished("这次起草已经结束了，不需要停止")
|
||||
if job.status == "queued":
|
||||
now = utcnow()
|
||||
claimed = await session.execute(
|
||||
update(CatalogDraftJob).where(CatalogDraftJob.id == job.id, CatalogDraftJob.status == "queued")
|
||||
.values(status="cancelled", finished_at=now).execution_options(synchronize_session=False))
|
||||
if claimed.rowcount == 1:
|
||||
await session.execute(
|
||||
update(CatalogDraftTask).where(CatalogDraftTask.job_id == job.id, CatalogDraftTask.status.in_(_OPEN))
|
||||
.values(status="cancelled", finished_at=now).execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
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 _close(session: AsyncSession, job_id: str, job_status: str, job_error: str | None, task_status: str,
|
||||
task_error: str | None) -> None:
|
||||
"""收尾：还没结束的表记 task_status，任务（还没结束的话）记 job_status。"""
|
||||
now = utcnow()
|
||||
await session.execute(
|
||||
update(CatalogDraftTask).where(CatalogDraftTask.job_id == job_id, CatalogDraftTask.status.in_(_OPEN))
|
||||
.values(status=task_status, error=task_error, finished_at=now).execution_options(synchronize_session=False))
|
||||
await session.execute(
|
||||
update(CatalogDraftJob).where(CatalogDraftJob.id == job_id, CatalogDraftJob.status.in_(ACTIVE))
|
||||
.values(status=job_status, error=job_error, finished_at=now).execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
|
||||
async def startup() -> int:
|
||||
"""服务启动时：排队、进行中的任务标成 interrupted，没结束的表记 cancelled、写明服务重启。不自动续跑。返回标了几个。"""
|
||||
async with SessionLocal() as session:
|
||||
ids = [job_id for job_id in (await session.execute(
|
||||
select(CatalogDraftJob.id).where(CatalogDraftJob.status.in_(ACTIVE)))).scalars()
|
||||
if not manager.running(job_id)]
|
||||
for job_id in ids:
|
||||
await _close(session, job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR)
|
||||
if ids:
|
||||
logger.warning("服务重启：%d 个数据目录起草任务没有做完，已标为中断，可在数据目录页重试", len(ids))
|
||||
return len(ids)
|
||||
|
||||
|
||||
# ==========================================================================
|
||||
# 在后台起草
|
||||
# ==========================================================================
|
||||
|
||||
|
||||
class _JobRun:
|
||||
"""一个任务在后台的一次执行。"""
|
||||
|
||||
def __init__(self, job_id: str) -> None:
|
||||
self.job_id = job_id
|
||||
self.task: asyncio.Task[None] | None = None
|
||||
#: 为什么停：user 使用者点了停止，shutdown 服务关停（标中断）
|
||||
self.stop_reason: str | None = None
|
||||
self._writes: set[asyncio.Task[Any]] = set()
|
||||
self._lock = asyncio.Lock()
|
||||
self._started: set[str] = set()
|
||||
self.source: Any = None
|
||||
self.actor: str | None = None
|
||||
self.at = ""
|
||||
|
||||
# ---------------------------------------------------------------- 库写入
|
||||
|
||||
async def _write(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> Any:
|
||||
"""一条库写入：同一个任务的写入一条一条来；shield 起来，停止任务时不会打断写到一半的库操作。"""
|
||||
async def go() -> Any:
|
||||
async with self._lock:
|
||||
async with SessionLocal() as session:
|
||||
return await fn(session, *args)
|
||||
|
||||
task = asyncio.ensure_future(go())
|
||||
self._writes.add(task)
|
||||
task.add_done_callback(self._writes.discard)
|
||||
return await asyncio.shield(task)
|
||||
|
||||
async def _settle(self) -> None:
|
||||
"""等手上的库写入都写完（停止、出错收尾之前）。"""
|
||||
while self._writes:
|
||||
await asyncio.gather(*list(self._writes), return_exceptions=True)
|
||||
|
||||
# ---------------------------------------------------------------- 主流程
|
||||
|
||||
async def run(self) -> None:
|
||||
try:
|
||||
await self._run()
|
||||
except asyncio.CancelledError:
|
||||
current = asyncio.current_task()
|
||||
if current is not None:
|
||||
current.uncancel()
|
||||
await self._settle()
|
||||
if self.stop_reason == "user":
|
||||
await self._write(_close, self.job_id, "cancelled", None, "cancelled", None)
|
||||
else:
|
||||
await self._write(_close, self.job_id, "interrupted", INTERRUPTED_ERROR, "cancelled", INTERRUPTED_ERROR)
|
||||
except _JobFailed as e:
|
||||
await self._settle()
|
||||
await self._write(_close, self.job_id, "failed", str(e), "failed", str(e))
|
||||
except Exception as e: # noqa: BLE001 - 任务整体出错：照实记下，不留在「起草中」
|
||||
from app.core.errors import describe_exception, first_line
|
||||
|
||||
logger.exception("数据目录起草任务 %s 出错", self.job_id)
|
||||
await self._settle()
|
||||
reason = JOB_ERROR.format(line=first_line(e) or describe_exception(e))
|
||||
await self._write(_close, self.job_id, "failed", reason, "failed", reason)
|
||||
|
||||
async def _run(self) -> None:
|
||||
if not await self._write(self._claim):
|
||||
return # 开始之前就被停止了
|
||||
async with SessionLocal() as session:
|
||||
job = await session.get(CatalogDraftJob, 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 _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 _JobFailed(str(e)) from e
|
||||
pending = [t.table_name for t in await tasks_of(session, job.id) if t.status == "pending"]
|
||||
use_model, background = bool(job.use_model), catalog_draft.Background(guidance=job.guidance, note=job.note)
|
||||
self.actor = job.created_by
|
||||
model: Any = None
|
||||
existing: dict[str, Any] = {}
|
||||
if use_model and pending:
|
||||
try:
|
||||
model, model_id = await catalog.resolve_draft_model(session)
|
||||
except catalog.CatalogModelUnavailable as e:
|
||||
raise _JobFailed(str(e)) from e
|
||||
existing = {t: e.notes for t, e in (await catalog.read_catalog(session, source.id)).items()}
|
||||
self.source, self.at = source, catalog.now_iso()
|
||||
all_tables = (getattr(source, "schema_cache", None) or {}).get("tables") or {}
|
||||
names = [n for n in pending if isinstance(all_tables.get(n), dict)]
|
||||
gone = [n for n in pending if n not in names]
|
||||
if gone:
|
||||
# 建任务之后表结构变了（重新探查过）：这几张表跳过
|
||||
await self._write(self._skip, gone)
|
||||
if use_model and names:
|
||||
await self._write(self._set_model, model_id)
|
||||
hooks = catalog_draft.Hooks(call_started=self._call_started, usage=self._usage,
|
||||
part_done=self._part_done, table_done=self._table_done)
|
||||
await catalog_draft.run_model_draft(model, source, names, existing=existing, background=background,
|
||||
hooks=hooks)
|
||||
else:
|
||||
for name in names:
|
||||
await self._call_started([name])
|
||||
await self._table_done(name, None)
|
||||
await self._settle()
|
||||
await self._write(self._finish)
|
||||
|
||||
# ---------------------------------------------------------------- 进度回调（引擎调）
|
||||
|
||||
async def _call_started(self, tables: list[str]) -> None:
|
||||
fresh = [t for t in dict.fromkeys(tables) if t not in self._started]
|
||||
if fresh:
|
||||
self._started.update(fresh)
|
||||
await self._write(self._mark_running, fresh)
|
||||
|
||||
async def _usage(self, input_tokens: int, output_tokens: int) -> None:
|
||||
await self._write(self._add_usage, input_tokens, output_tokens)
|
||||
|
||||
async def _part_done(self, table: str) -> None:
|
||||
await self._write(self._bump_part, table)
|
||||
|
||||
async def _table_done(self, table: str, outcome: catalog_draft.ModelOutcome | None) -> None:
|
||||
await self._write(self._write_table, table, outcome)
|
||||
|
||||
# ---------------------------------------------------------------- 库操作
|
||||
|
||||
def _task(self, *where: Any) -> Any:
|
||||
return update(CatalogDraftTask).where(CatalogDraftTask.job_id == self.job_id, *where) \
|
||||
.execution_options(synchronize_session=False)
|
||||
|
||||
async def _claim(self, session: AsyncSession) -> bool:
|
||||
result = await session.execute(
|
||||
update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id, CatalogDraftJob.status == "queued")
|
||||
.values(status="running", started_at=utcnow()).execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
return result.rowcount == 1
|
||||
|
||||
async def _set_model(self, session: AsyncSession, model_id: str) -> None:
|
||||
await session.execute(update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id)
|
||||
.values(model=model_id or None).execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
async def _skip(self, session: AsyncSession, names: list[str]) -> None:
|
||||
now = utcnow()
|
||||
for name in names:
|
||||
await session.execute(self._task(CatalogDraftTask.table_name == name, CatalogDraftTask.status == "pending")
|
||||
.values(status="skipped", error=catalog.missing_table_error(name), finished_at=now))
|
||||
await session.commit()
|
||||
|
||||
async def _mark_running(self, session: AsyncSession, names: list[str]) -> None:
|
||||
await session.execute(self._task(CatalogDraftTask.table_name.in_(names), CatalogDraftTask.status == "pending")
|
||||
.values(status="running", started_at=utcnow()))
|
||||
await session.commit()
|
||||
|
||||
async def _add_usage(self, session: AsyncSession, input_tokens: int, output_tokens: int) -> None:
|
||||
await session.execute(
|
||||
update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id)
|
||||
.values(calls=CatalogDraftJob.calls + 1, input_tokens=CatalogDraftJob.input_tokens + int(input_tokens),
|
||||
output_tokens=CatalogDraftJob.output_tokens + int(output_tokens))
|
||||
.execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
async def _bump_part(self, session: AsyncSession, table: str) -> None:
|
||||
await session.execute(self._task(CatalogDraftTask.table_name == table)
|
||||
.values(parts_done=CatalogDraftTask.parts_done + 1))
|
||||
await session.commit()
|
||||
|
||||
async def _write_table(self, session: AsyncSession, table: str,
|
||||
outcome: catalog_draft.ModelOutcome | None) -> None:
|
||||
"""一张表结束：并入目录、写入，记下这张表的结果。"""
|
||||
try:
|
||||
result = await catalog.write_draft(session, self.source, table, outcome, at=self.at, actor=self.actor)
|
||||
except Exception as e: # noqa: BLE001 - 这一张写不进去不影响别的表
|
||||
from app.core.errors import describe_exception, first_line
|
||||
|
||||
logger.exception("起草任务 %s 写入表 %s 的数据目录时出错", self.job_id, table)
|
||||
await session.rollback()
|
||||
result = catalog.TableDraftResult(table=table, status="failed",
|
||||
error=WRITE_ERROR.format(line=first_line(e) or describe_exception(e)),
|
||||
model_error=outcome.error if outcome is not None else None)
|
||||
values: dict[str, Any] = {
|
||||
"status": result.status, "added": result.added, "updated": result.updated, "removed": result.removed,
|
||||
"version": result.version, "model_items": result.model_items, "error": result.error,
|
||||
"model_error": result.model_error, "finished_at": utcnow(),
|
||||
}
|
||||
if outcome is not None:
|
||||
values["parts_done"] = outcome.parts_done
|
||||
await session.execute(self._task(CatalogDraftTask.table_name == table, CatalogDraftTask.status.in_(_OPEN))
|
||||
.values(**values))
|
||||
await session.commit()
|
||||
|
||||
async def _finish(self, session: AsyncSession) -> None:
|
||||
"""起草完：万一还有没收尾的表（进度回调写库失败之类）记失败；有 failed、partial、cancelled 的表是 partial，
|
||||
否则 succeeded。"""
|
||||
now = utcnow()
|
||||
await session.execute(self._task(CatalogDraftTask.status.in_(_OPEN))
|
||||
.values(status="failed", error=UNFINISHED_ERROR, finished_at=now))
|
||||
statuses = list((await session.execute(
|
||||
select(CatalogDraftTask.status).where(CatalogDraftTask.job_id == self.job_id))).scalars())
|
||||
status = "partial" if any(s in RETRYABLE for s in statuses) else "succeeded"
|
||||
await session.execute(
|
||||
update(CatalogDraftJob).where(CatalogDraftJob.id == self.job_id, CatalogDraftJob.status == "running")
|
||||
.values(status=status, finished_at=now).execution_options(synchronize_session=False))
|
||||
await session.commit()
|
||||
|
||||
|
||||
class DraftJobs:
|
||||
"""进程内的起草任务（照 app/scheduler 的写法：一个 asyncio 任务一件事，停的时候给宽限）。"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self._runs: dict[str, _JobRun] = {}
|
||||
|
||||
def _alive(self, job_id: str) -> _JobRun | None:
|
||||
run = self._runs.get(job_id)
|
||||
if run is None or run.task is None or run.task.done():
|
||||
return None
|
||||
try:
|
||||
if run.task.get_loop() is not asyncio.get_running_loop():
|
||||
self._runs.pop(job_id, None) # 别的事件循环留下的（测试每个用例一个循环）
|
||||
return None
|
||||
except RuntimeError:
|
||||
return None
|
||||
return run
|
||||
|
||||
def start(self, job_id: str) -> None:
|
||||
"""开始在后台起草这个（排队中的）任务。"""
|
||||
run = _JobRun(job_id)
|
||||
run.task = asyncio.get_running_loop().create_task(run.run(), name=f"catalog-draft-{job_id}")
|
||||
self._runs[job_id] = run
|
||||
|
||||
def forget(_: asyncio.Task[None]) -> None:
|
||||
if self._runs.get(job_id) is run:
|
||||
self._runs.pop(job_id, None)
|
||||
|
||||
run.task.add_done_callback(forget)
|
||||
|
||||
def running(self, job_id: str) -> bool:
|
||||
return self._alive(job_id) is not None
|
||||
|
||||
async def stop(self, job_id: str, reason: str, *, grace: float = STOP_GRACE_S) -> bool:
|
||||
"""停掉这个进程里在跑的任务，等它收尾（最多 grace 秒）。不在这个进程里跑返回 False。"""
|
||||
run = self._alive(job_id)
|
||||
if run is None:
|
||||
return False
|
||||
if run.stop_reason is None:
|
||||
run.stop_reason = reason
|
||||
run.task.cancel() # type: ignore[union-attr]
|
||||
await asyncio.wait([run.task], timeout=grace) # type: ignore[list-item]
|
||||
return True
|
||||
|
||||
async def wait(self, job_id: str, timeout: float | None = None) -> None:
|
||||
"""等这个任务做完（测试、调试用）。不在跑直接返回。"""
|
||||
run = self._alive(job_id)
|
||||
if run is not None:
|
||||
await asyncio.wait([run.task], timeout=timeout) # type: ignore[list-item]
|
||||
|
||||
async def shutdown(self, *, grace: float = 5.0) -> None:
|
||||
"""关停：停掉进行中的任务（标中断），最多等 grace 秒收尾。"""
|
||||
runs = [run for job_id in list(self._runs) if (run := self._alive(job_id)) is not None]
|
||||
for run in runs:
|
||||
if run.stop_reason is None:
|
||||
run.stop_reason = "shutdown"
|
||||
run.task.cancel() # type: ignore[union-attr]
|
||||
if runs:
|
||||
await asyncio.wait([run.task for run in runs], timeout=grace) # type: ignore[misc]
|
||||
logger.info("已停止 %d 个数据目录起草任务", len(runs))
|
||||
|
||||
|
||||
manager = DraftJobs()
|
||||
|
||||
|
||||
__all__ = ["ACTIVE", "DraftBusy", "DraftFinished", "DraftJobs", "FINISHED", "GUIDANCE_MAX", "INTERRUPTED_ERROR",
|
||||
"MAX_TABLES", "NOTE_MAX", "NothingToRetry", "RETRYABLE", "TASK_DONE", "TASK_STATUSES", "TextTooLong",
|
||||
"active_job", "cancel", "check_model", "clean_text", "create_job", "ensure_idle", "guidance_of", "job_brief",
|
||||
"job_dict", "job_out", "last_drafts", "latest_job", "load", "manager", "retry_tables", "save_guidance",
|
||||
"startup", "tasks_of"]
|
||||
@@ -62,12 +62,18 @@ MASK_COLUMNS_OPTION = "mask_columns"
|
||||
#: data/catalog_profile.py；放在这里只为让拼连接串时摘掉它
|
||||
CATALOG_PROFILE_OPTION = "catalog_profile"
|
||||
|
||||
#: 数据源 options 里的数据目录起草要求：使用者补充的背景和要求，起草时交给模型（一段文字，最多 2000 字）。
|
||||
#: 只经 PUT /api/datasources/{id}/catalog/guidance 维护，见 data/catalog_jobs.py；放在这里只为让拼连接串时摘掉它
|
||||
CATALOG_DRAFT_GUIDANCE_OPTION = "catalog_draft_guidance"
|
||||
|
||||
# options 里这些 key 是 AgentLab 自己的配置，不是驱动参数，拼 URL 时要摘掉。
|
||||
# schema：探查哪个 schema（企业库里只读账号名下常常什么都没有，数据在别处）
|
||||
# query_timeout_s：查询时限，由数据层按语句下发给数据库，见 _server_deadline
|
||||
# mask_columns：证据面板遮罩的列，驱动不认识它，拼进连接串会被当成未知参数拒掉
|
||||
# catalog_profile：数据剖析设置，同上；它还是个对象，拼进去连 URL 都不成形
|
||||
_NON_DRIVER_OPTIONS = frozenset({"schema", QUERY_TIMEOUT_OPTION, MASK_COLUMNS_OPTION, CATALOG_PROFILE_OPTION})
|
||||
# catalog_draft_guidance：数据目录的起草要求，一段中文，拼进去连接直接失败
|
||||
_NON_DRIVER_OPTIONS = frozenset({"schema", QUERY_TIMEOUT_OPTION, MASK_COLUMNS_OPTION, CATALOG_PROFILE_OPTION,
|
||||
CATALOG_DRAFT_GUIDANCE_OPTION})
|
||||
|
||||
_MASK_SPLIT = re.compile(r"[,，、;；\n]+")
|
||||
|
||||
|
||||
@@ -4,7 +4,7 @@ from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import (
|
||||
Float, ForeignKey, Index, Integer, LargeBinary, String, Text, UniqueConstraint,
|
||||
Float, ForeignKey, Index, Integer, LargeBinary, String, Text, UniqueConstraint, text,
|
||||
)
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
@@ -348,6 +348,76 @@ class CatalogNote(Base, TimestampMixin):
|
||||
updated_by: Mapped[str | None] = mapped_column(String(100), default=None)
|
||||
|
||||
|
||||
class CatalogDraftJob(Base, TimestampMixin):
|
||||
"""数据目录的一次后台起草任务（data/catalog_jobs.py）。逐表的状态在 catalog_draft_tasks。
|
||||
|
||||
以前起草是一次同步长请求，出错整个对话框就停、关掉之后没有任何记录哪些表失败、为什么；现在建任务后在后台起草，
|
||||
进度（每张表到第几段、计数、用量）随时落库，接口读的是库里的状态。
|
||||
|
||||
同一个数据源同时只有一个排队或进行中的任务：接口先查一遍，部分唯一索引兜底（两个请求同时建，后提交的撞上）。
|
||||
服务重启时还没结束的标成 interrupted，不自动续跑。挂外键、级联删除：任务是数据源的附属记录，源删掉了留着没有意义。
|
||||
"""
|
||||
|
||||
__tablename__ = "catalog_draft_jobs"
|
||||
__table_args__ = (
|
||||
Index("ix_catalog_draft_jobs_source_created", "source_id", "created_at"),
|
||||
Index("ux_catalog_draft_jobs_active", "source_id", unique=True,
|
||||
sqlite_where=text("status IN ('queued', 'running')")),
|
||||
)
|
||||
|
||||
id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
|
||||
source_id: Mapped[str] = mapped_column(ForeignKey("data_sources.id", ondelete="CASCADE"))
|
||||
# queued | running | succeeded | partial | failed | cancelled | interrupted
|
||||
status: Mapped[str] = mapped_column(String(20), default="queued")
|
||||
use_model: Mapped[bool] = mapped_column(default=False)
|
||||
# 实际用的模型 id（用模型时开始起草后写上）
|
||||
model: Mapped[str | None] = mapped_column(String(200), default=None)
|
||||
# 建任务时的快照：数据源级的起草要求、这一次的补充说明
|
||||
guidance: Mapped[str | None] = mapped_column(Text, default=None)
|
||||
note: Mapped[str | None] = mapped_column(Text, default=None)
|
||||
# 请求头 X-Actor 的署名（自报、未认证）
|
||||
created_by: Mapped[str | None] = mapped_column(String(100), default=None)
|
||||
started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
|
||||
finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
|
||||
# 用量：调用次数、输入和输出 token（拿不到用量的调用按 0 算）
|
||||
calls: Mapped[int] = mapped_column(Integer, default=0)
|
||||
input_tokens: Mapped[int] = mapped_column(Integer, default=0)
|
||||
output_tokens: Mapped[int] = mapped_column(Integer, default=0)
|
||||
# 任务整体出错的原因（逐表的原因在 catalog_draft_tasks）
|
||||
error: Mapped[str | None] = mapped_column(Text, default=None)
|
||||
# 重试建的任务记下是重试哪一个
|
||||
retry_of: Mapped[str | None] = mapped_column(String(32), default=None)
|
||||
|
||||
|
||||
class CatalogDraftTask(Base):
|
||||
"""起草任务里的一张表。按 seq（起草顺序）排。状态：pending 排队、running 起草中、ok、partial 模型只写出一部分、
|
||||
failed、skipped 表结构里没有、cancelled（停止或服务重启）。"""
|
||||
|
||||
__tablename__ = "catalog_draft_tasks"
|
||||
__table_args__ = (UniqueConstraint("job_id", "seq"),)
|
||||
|
||||
id: Mapped[str] = mapped_column(String(32), primary_key=True, default=new_id)
|
||||
job_id: Mapped[str] = mapped_column(ForeignKey("catalog_draft_jobs.id", ondelete="CASCADE"), index=True)
|
||||
seq: Mapped[int] = mapped_column(Integer)
|
||||
# 表结构里的键；表结构里没有的（skipped）是请求里的原样名字
|
||||
table_name: Mapped[str] = mapped_column(String(255))
|
||||
status: Mapped[str] = mapped_column(String(20), default="pending")
|
||||
column_count: Mapped[int] = mapped_column(Integer, default=0)
|
||||
# 模型分几段起草（不用模型时为 NULL）、已经结束了几段
|
||||
parts_total: Mapped[int | None] = mapped_column(Integer, default=None)
|
||||
parts_done: Mapped[int] = mapped_column(Integer, default=0)
|
||||
added: Mapped[int] = mapped_column(Integer, default=0)
|
||||
updated: Mapped[int] = mapped_column(Integer, default=0)
|
||||
removed: Mapped[int] = mapped_column(Integer, default=0)
|
||||
version: Mapped[int] = mapped_column(Integer, default=0)
|
||||
model_items: Mapped[int | None] = mapped_column(Integer, default=None)
|
||||
# error：整张表没起草（表结构里没有、写入一直冲突、服务重启中断）；model_error：模型那部分没写出来的原因和办法
|
||||
error: Mapped[str | None] = mapped_column(Text, default=None)
|
||||
model_error: Mapped[str | None] = mapped_column(Text, default=None)
|
||||
started_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
|
||||
finished_at: Mapped[datetime | None] = mapped_column(UTCDateTime, default=None)
|
||||
|
||||
|
||||
class Setting(Base, TimestampMixin):
|
||||
"""键值形式的用户设置，前端 Settings 页直接读写。"""
|
||||
|
||||
|
||||
+11
-1
@@ -70,6 +70,15 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("上传表格的版本整理失败，服务照常启动")
|
||||
|
||||
# 上次没做完的数据目录起草任务：标成中断（没结束的表记停止），不自动续跑——拷来的数据目录（检查脚本、验收副本）
|
||||
# 启动时不能替人去调模型。界面上点「重试」接着起草没写好的表
|
||||
from app.data import catalog_jobs
|
||||
|
||||
try:
|
||||
await catalog_jobs.startup()
|
||||
except Exception: # noqa: BLE001
|
||||
logger.exception("数据目录起草任务的启动整理失败，服务照常启动")
|
||||
|
||||
await run_manager.setup()
|
||||
from app.sandbox.manager import sandbox_manager as _sbm
|
||||
|
||||
@@ -84,8 +93,9 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
# 先停调度，再停运行：停运行的途中不该再有新运行被发起
|
||||
# 先停调度，再停运行：停运行的途中不该再有新运行被发起。起草任务停下来标成中断，下次启动不会再续
|
||||
await scheduler.stop()
|
||||
await catalog_jobs.manager.shutdown()
|
||||
await run_manager.shutdown()
|
||||
from app.sandbox.manager import sandbox_manager
|
||||
|
||||
|
||||
@@ -0,0 +1,526 @@
|
||||
"""数据目录的后台起草任务（data/catalog_jobs.py、/api/datasources/{id}/catalog/draft-jobs）与起草要求。
|
||||
|
||||
- 建任务 202；同一个源已有排队或进行中的任务 409 catalog_draft_busy；起草模型用不了 409
|
||||
catalog_draft_model_unavailable（不建任务）；没有表结构 409；表结构里没有的表照样建进任务、状态 skipped
|
||||
- 在后台起草，每张表一结束就并入目录；进度、计数、用量落库，接口读的是库里的状态
|
||||
- 停止：排队的直接 cancelled；进行中的停掉正在进行的模型调用，没写完的表记 cancelled，已经写进目录的不回滚；已结束的 409
|
||||
- 服务启动时把没结束的任务标成 interrupted（没结束的表记 cancelled、写明服务重启），不自动续跑；关停时同样
|
||||
- 重试：只含原任务里 failed、partial、cancelled 的表，本次说明沿用、起草要求取数据源当前的；没有可重试的 422
|
||||
- 目录清单：每行 last_draft，顶层 guidance、active_job
|
||||
- 起草要求存在数据源 options 里（上传表格的源存在它自己那一行），数据源接口不显示、整份回写 options 时留着，拼连接串时摘掉
|
||||
|
||||
表名、列名、说明文字都是虚构的。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
from urllib.parse import quote
|
||||
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy import select
|
||||
|
||||
from app.data import catalog, catalog_jobs
|
||||
from app.data.engine import CATALOG_DRAFT_GUIDANCE_OPTION, build_url
|
||||
from app.db.base import SessionLocal
|
||||
from app.db.models import CatalogDraftJob, CatalogDraftTask, DataSource
|
||||
from app.main import app as _fastapi_app
|
||||
from tests.fixtures.catalog.fake_model import FakeModel, requested
|
||||
from tests.fixtures.sources import drop_source
|
||||
|
||||
ACTOR = {"X-Actor": quote("王敏")}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 夹具
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _col(name: str, type_: str = "INTEGER", comment: str | None = None) -> dict:
|
||||
return {"name": name, "type": type_, "nullable": True, "comment": comment}
|
||||
|
||||
|
||||
def _tables() -> dict:
|
||||
return {
|
||||
"visits": {"qualified": "visits", "columns": [_col("id"), _col("ticket_no", "TEXT"), _col("visit_time", "TEXT"),
|
||||
_col("channel_id")],
|
||||
"primary_key": ["id"], "comment": None, "is_view": False,
|
||||
"foreign_keys": [{"columns": ["channel_id"], "to_table": "channels", "to_columns": ["id"]}]},
|
||||
"channels": {"qualified": "channels", "columns": [_col("id"), _col("name", "TEXT")], "primary_key": ["id"],
|
||||
"comment": "售票渠道", "is_view": False, "foreign_keys": []},
|
||||
"gates": {"qualified": "gates", "columns": [_col("id"), _col("gate_code", "TEXT")], "primary_key": ["id"],
|
||||
"comment": None, "is_view": False, "foreign_keys": []},
|
||||
}
|
||||
|
||||
|
||||
REPLIES = {
|
||||
"visits": {"label": "入园记录", "kind": "fact", "columns": [{"name": "ticket_no", "label": "票号"}]},
|
||||
"channels": {"label": "渠道", "kind": "dimension"},
|
||||
"gates": {"label": "闸口", "kind": "dimension"},
|
||||
}
|
||||
|
||||
_CREATED: list[str] = []
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
async def _cleanup():
|
||||
yield
|
||||
# 后台任务先收完（停不下来的停掉），再删源：删源会级联删掉任务
|
||||
await catalog_jobs.manager.shutdown(grace=5)
|
||||
while _CREATED:
|
||||
await drop_source(_CREATED.pop())
|
||||
|
||||
|
||||
async def _source(tables: dict | None = None, **kw) -> str:
|
||||
async with SessionLocal() as session:
|
||||
row = DataSource(name=f"cdj_{uuid.uuid4().hex[:10]}", kind="sqlite", database=":memory:",
|
||||
schema_cache={"tables": tables if tables is not None else _tables()}, **kw)
|
||||
session.add(row)
|
||||
await session.commit()
|
||||
_CREATED.append(row.id)
|
||||
return row.id
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client():
|
||||
async with AsyncClient(transport=ASGITransport(app=_fastapi_app), base_url="http://test") as c:
|
||||
yield c
|
||||
|
||||
|
||||
def _base(sid: str) -> str:
|
||||
return f"/api/datasources/{sid}/catalog"
|
||||
|
||||
|
||||
def use(monkeypatch, model) -> None:
|
||||
"""起草模型换成假模型（建任务时查一次能不能用，开始起草时再取一次）。"""
|
||||
async def resolve(session):
|
||||
return model, "fake-model"
|
||||
|
||||
monkeypatch.setattr(catalog, "resolve_draft_model", resolve)
|
||||
|
||||
|
||||
async def finished(job_id: str) -> None:
|
||||
await catalog_jobs.manager.wait(job_id, timeout=10)
|
||||
|
||||
|
||||
async def get_job(client, sid: str, job_id: str) -> dict:
|
||||
r = await client.get(f"{_base(sid)}/draft-jobs/{job_id}")
|
||||
assert r.status_code == 200, r.text
|
||||
return r.json()["job"]
|
||||
|
||||
|
||||
class Held:
|
||||
"""假模型：请求里有 hold 里的表时，等 release 才回；被停止时记下这次调用被取消了。"""
|
||||
|
||||
def __init__(self, replies: dict, hold: set[str]) -> None:
|
||||
self.inner, self.hold = FakeModel(replies), set(hold)
|
||||
self.release = asyncio.Event()
|
||||
self.cancelled = 0
|
||||
self.model_name = "held-model"
|
||||
|
||||
def with_structured_output(self, schema, **_):
|
||||
outer = self
|
||||
|
||||
class _S:
|
||||
async def ainvoke(self, messages):
|
||||
text = "\n".join(str(m.content) for m in messages)
|
||||
if outer.hold & set(requested(text)):
|
||||
try:
|
||||
await outer.release.wait()
|
||||
except asyncio.CancelledError:
|
||||
outer.cancelled += 1
|
||||
raise
|
||||
return outer.inner._reply(messages, "structured")
|
||||
|
||||
return _S()
|
||||
|
||||
async def ainvoke(self, messages):
|
||||
return await self.inner.ainvoke(messages)
|
||||
|
||||
|
||||
async def until(predicate, timeout: float = 5.0) -> None:
|
||||
for _ in range(int(timeout / 0.01)):
|
||||
if await predicate():
|
||||
return
|
||||
await asyncio.sleep(0.01)
|
||||
raise AssertionError("等不到")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 建任务、在后台起草
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_a_job_drafts_in_the_background_and_writes_each_table(client, monkeypatch):
|
||||
sid = await _source()
|
||||
model = FakeModel(REPLIES)
|
||||
use(monkeypatch, model)
|
||||
await client.put(f"{_base(sid)}/guidance", json={"guidance": "金额一律按元写单位"})
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs", headers=ACTOR,
|
||||
json={"tables": ["visits", "nope", "channels"], "use_model": True, "note": " 看渠道 "})
|
||||
assert r.status_code == 202, r.text
|
||||
job = r.json()["job"]
|
||||
assert job["status"] in ("queued", "running") and job["source_id"] == sid
|
||||
assert job["use_model"] is True and job["guidance"] == "金额一律按元写单位" and job["note"] == "看渠道"
|
||||
assert job["created_by"] == "王敏" and job["total"] == 3
|
||||
assert [t["table_name"] for t in job["tasks"]] == ["visits", "nope", "channels"], "按起草顺序"
|
||||
await finished(job["id"])
|
||||
|
||||
job = await get_job(client, sid, job["id"])
|
||||
assert job["status"] == "succeeded" and job["error"] is None and job["model"] == "fake-model"
|
||||
assert job["done"] == 3 and job["counts"] == {"pending": 0, "running": 0, "ok": 2, "partial": 0, "failed": 0,
|
||||
"skipped": 1, "cancelled": 0}
|
||||
assert job["started_at"] and job["finished_at"] and job["usage"]["calls"] >= 1
|
||||
by = {t["table_name"]: t for t in job["tasks"]}
|
||||
assert by["nope"]["status"] == "skipped" and by["nope"]["error"] == "表结构里没有 nope，请先重新探查结构"
|
||||
visits = by["visits"]
|
||||
assert visits["status"] == "ok" and visits["parts_total"] == 1 and visits["parts_done"] == 1
|
||||
assert visits["model_items"] == 3 and visits["added"] == 4 and visits["version"] == 1 and visits["columns"] == 4
|
||||
assert visits["started_at"] and visits["finished_at"] and visits["error"] is None
|
||||
assert job["added"] == sum(t["added"] for t in job["tasks"])
|
||||
# 起草要求和本次说明都交给了模型
|
||||
assert "起草要求：金额一律按元写单位" in model.calls[0] and "本次说明：看渠道" in model.calls[0]
|
||||
|
||||
detail = (await client.get(f"{_base(sid)}/visits")).json()
|
||||
assert detail["notes"]["label"]["value"] == "入园记录" and detail["updated_by"] == "王敏"
|
||||
|
||||
|
||||
async def test_a_job_without_the_model_writes_structure_only(client):
|
||||
sid = await _source()
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": [], "use_model": False})
|
||||
assert r.status_code == 202, r.text
|
||||
job = r.json()["job"]
|
||||
assert job["total"] == 3, "不给表时按使用次数取前 20 张"
|
||||
await finished(job["id"])
|
||||
job = await get_job(client, sid, job["id"])
|
||||
assert job["status"] == "succeeded" and job["model"] is None and job["usage"]["calls"] == 0
|
||||
assert all(t["parts_total"] is None and t["model_items"] is None for t in job["tasks"])
|
||||
assert {t["table_name"]: t["added"] for t in job["tasks"]}["visits"] == 1 # 外键一条
|
||||
|
||||
|
||||
async def test_a_table_whose_model_part_failed_is_failed_but_structure_is_written(client, monkeypatch):
|
||||
sid = await _source()
|
||||
use(monkeypatch, FakeModel(REPLIES, fail={"visits"}))
|
||||
monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1)
|
||||
job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits", "channels"],
|
||||
"use_model": True})).json()["job"]
|
||||
await finished(job["id"])
|
||||
job = await get_job(client, sid, job["id"])
|
||||
by = {t["table_name"]: t for t in job["tasks"]}
|
||||
assert job["status"] == "partial"
|
||||
assert by["visits"]["status"] == "failed" and "网关超时" in by["visits"]["model_error"]
|
||||
assert by["visits"]["error"] is None and by["visits"]["added"] == 1, "外键照常写入"
|
||||
assert by["channels"]["status"] == "ok"
|
||||
|
||||
|
||||
async def test_busy_source_is_409_and_the_active_job_is_reachable(client, monkeypatch):
|
||||
sid = await _source()
|
||||
model = Held(REPLIES, hold={"visits"})
|
||||
use(monkeypatch, model)
|
||||
first = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits"], "use_model": True})).json()
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"], "use_model": False})
|
||||
assert r.status_code == 409 and r.json()["code"] == "catalog_draft_busy", r.text
|
||||
active = (await client.get(f"{_base(sid)}/draft-jobs/active")).json()["job"]
|
||||
assert active["id"] == first["job"]["id"]
|
||||
listed = (await client.get(_base(sid))).json()
|
||||
assert listed["active_job"]["id"] == first["job"]["id"] and listed["active_job"]["total"] == 1
|
||||
assert listed["active_job"]["status"] in ("queued", "running") and listed["active_job"]["done"] == 0
|
||||
model.release.set()
|
||||
await finished(first["job"]["id"])
|
||||
assert (await client.get(f"{_base(sid)}/draft-jobs/active")).json()["job"] is None
|
||||
latest = (await client.get(f"{_base(sid)}/draft-jobs/latest")).json()["job"]
|
||||
assert latest["id"] == first["job"]["id"] and latest["status"] == "succeeded"
|
||||
assert (await client.get(_base(sid))).json()["active_job"] is None
|
||||
|
||||
|
||||
async def test_unavailable_model_is_409_and_no_job_is_created(client, monkeypatch):
|
||||
sid = await _source()
|
||||
|
||||
async def unavailable(session):
|
||||
raise catalog.CatalogModelUnavailable("未配置模型接入，无法由模型起草数据目录。请到「设置 → 模型接入」添加")
|
||||
|
||||
monkeypatch.setattr(catalog, "resolve_draft_model", unavailable)
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs", json={"use_model": True})
|
||||
assert r.status_code == 409, r.text
|
||||
assert r.json()["code"] == "catalog_draft_model_unavailable" and "模型接入" in r.json()["detail"]
|
||||
assert (await client.get(f"{_base(sid)}/draft-jobs/latest")).json()["job"] is None
|
||||
|
||||
|
||||
async def test_without_schema_or_with_bad_input_nothing_is_created(client):
|
||||
sid = await _source(tables={})
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs", json={"use_model": False})
|
||||
assert r.status_code == 409 and "探查结构" in r.json()["detail"]
|
||||
sid = await _source()
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs", json={"use_model": False, "note": "说" * 1001})
|
||||
assert r.status_code == 422 and "1000" in r.json()["detail"]
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs", json={"tables": [f"t{i}" for i in range(501)]})
|
||||
assert r.status_code == 422 and "500" in r.json()["detail"]
|
||||
assert (await client.get(f"{_base(sid)}/draft-jobs/latest")).json()["job"] is None
|
||||
assert (await client.get(f"{_base('missing-source')}/draft-jobs/latest")).status_code == 404
|
||||
assert (await client.get(f"{_base(sid)}/draft-jobs/not-a-job")).status_code == 404
|
||||
|
||||
|
||||
async def test_a_job_of_another_source_is_404(client):
|
||||
a, b = await _source(), await _source()
|
||||
job = (await client.post(f"{_base(a)}/draft-jobs", json={"tables": ["gates"]})).json()["job"]
|
||||
await finished(job["id"])
|
||||
assert (await client.get(f"{_base(b)}/draft-jobs/{job['id']}")).status_code == 404
|
||||
assert (await client.post(f"{_base(b)}/draft-jobs/{job['id']}/cancel")).status_code == 404
|
||||
assert (await client.post(f"{_base(b)}/draft-jobs/{job['id']}/retry")).status_code == 404
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 停止、服务重启
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_cancelling_a_queued_job(client, monkeypatch):
|
||||
sid = await _source()
|
||||
monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None) # 不让它开始
|
||||
job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates", "channels"]})).json()["job"]
|
||||
assert job["status"] == "queued"
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel")
|
||||
assert r.status_code == 200, r.text
|
||||
job = r.json()["job"]
|
||||
assert job["status"] == "cancelled" and job["finished_at"]
|
||||
assert {t["status"] for t in job["tasks"]} == {"cancelled"}
|
||||
again = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel")
|
||||
assert again.status_code == 409 and again.json()["code"] == "catalog_draft_finished"
|
||||
|
||||
|
||||
async def test_cancelling_a_running_job_stops_the_call_and_keeps_what_is_written(client, monkeypatch):
|
||||
sid = await _source()
|
||||
model = Held(REPLIES, hold={"visits"})
|
||||
use(monkeypatch, model)
|
||||
monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1)
|
||||
job = (await client.post(f"{_base(sid)}/draft-jobs",
|
||||
json={"tables": ["channels", "visits"], "use_model": True})).json()["job"]
|
||||
|
||||
async def channels_written() -> bool:
|
||||
tasks = {t["table_name"]: t for t in (await get_job(client, sid, job["id"]))["tasks"]}
|
||||
return tasks["channels"]["status"] == "ok" and tasks["visits"]["status"] == "running"
|
||||
|
||||
await until(channels_written)
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel")
|
||||
assert r.status_code == 200, r.text
|
||||
after = r.json()["job"]
|
||||
by = {t["table_name"]: t for t in after["tasks"]}
|
||||
assert after["status"] == "cancelled" and after["finished_at"]
|
||||
assert by["channels"]["status"] == "ok" and by["visits"]["status"] == "cancelled"
|
||||
assert model.cancelled == 1, "正在进行的模型调用被停掉"
|
||||
detail = (await client.get(f"{_base(sid)}/channels")).json()
|
||||
assert detail["notes"]["kind"]["value"] == "dimension", "已经写进目录的不回滚"
|
||||
assert (await client.get(f"{_base(sid)}/draft-jobs/active")).json()["job"] is None
|
||||
|
||||
|
||||
async def test_startup_marks_unfinished_jobs_interrupted_and_does_not_resume(client):
|
||||
sid = await _source()
|
||||
async with SessionLocal() as session:
|
||||
job = CatalogDraftJob(source_id=sid, status="running", use_model=True)
|
||||
session.add(job)
|
||||
await session.flush()
|
||||
for seq, (name, status) in enumerate([("visits", "ok"), ("channels", "running"), ("gates", "pending")]):
|
||||
session.add(CatalogDraftTask(job_id=job.id, seq=seq, table_name=name, status=status))
|
||||
await session.commit()
|
||||
job_id = job.id
|
||||
assert await catalog_jobs.startup() >= 1
|
||||
got = await get_job(client, sid, job_id)
|
||||
by = {t["table_name"]: t for t in got["tasks"]}
|
||||
assert got["status"] == "interrupted" and got["error"] == "服务重启，起草中断" and got["finished_at"]
|
||||
assert by["visits"]["status"] == "ok" and by["visits"]["error"] is None
|
||||
assert by["channels"]["status"] == by["gates"]["status"] == "cancelled"
|
||||
assert by["channels"]["error"] == by["gates"]["error"] == "服务重启，起草中断"
|
||||
assert not catalog_jobs.manager.running(job_id), "不自动续跑"
|
||||
|
||||
|
||||
async def test_shutdown_stops_running_jobs_as_interrupted(client, monkeypatch):
|
||||
sid = await _source()
|
||||
model = Held(REPLIES, hold={"visits"})
|
||||
use(monkeypatch, model)
|
||||
job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits"], "use_model": True})).json()["job"]
|
||||
|
||||
async def running() -> bool:
|
||||
return (await get_job(client, sid, job["id"]))["tasks"][0]["status"] == "running"
|
||||
|
||||
await until(running)
|
||||
await catalog_jobs.manager.shutdown(grace=5)
|
||||
got = await get_job(client, sid, job["id"])
|
||||
assert got["status"] == "interrupted" and got["tasks"][0]["status"] == "cancelled"
|
||||
assert got["tasks"][0]["error"] == "服务重启，起草中断"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 重试
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_retry_takes_only_failed_partial_and_cancelled_tables(client, monkeypatch):
|
||||
sid = await _source()
|
||||
use(monkeypatch, FakeModel(REPLIES, fail={"visits"}))
|
||||
monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1)
|
||||
await client.put(f"{_base(sid)}/guidance", json={"guidance": "旧的要求"})
|
||||
first = (await client.post(f"{_base(sid)}/draft-jobs", json={
|
||||
"tables": ["visits", "channels", "nope"], "use_model": True, "note": "第一次"})).json()["job"]
|
||||
await finished(first["id"])
|
||||
await client.put(f"{_base(sid)}/guidance", json={"guidance": "新的要求"})
|
||||
assert (await get_job(client, sid, first["id"]))["guidance"] == "旧的要求", "任务记下的是当时的快照"
|
||||
|
||||
use(monkeypatch, FakeModel(REPLIES))
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs/{first['id']}/retry", headers=ACTOR)
|
||||
assert r.status_code == 202, r.text
|
||||
retry = r.json()["job"]
|
||||
assert [t["table_name"] for t in retry["tasks"]] == ["visits"]
|
||||
assert retry["use_model"] is True and retry["note"] == "第一次" and retry["guidance"] == "新的要求"
|
||||
assert retry["retry_of"] == first["id"] and retry["created_by"] == "王敏"
|
||||
await finished(retry["id"])
|
||||
assert (await get_job(client, sid, retry["id"]))["status"] == "succeeded"
|
||||
|
||||
nothing = await client.post(f"{_base(sid)}/draft-jobs/{retry['id']}/retry")
|
||||
assert nothing.status_code == 422 and "没有" in nothing.json()["detail"]
|
||||
|
||||
|
||||
async def test_retry_while_busy_is_409(client, monkeypatch):
|
||||
sid = await _source()
|
||||
monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None)
|
||||
job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"]})).json()["job"]
|
||||
await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel")
|
||||
other = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["channels"]})).json()["job"]
|
||||
assert other["status"] == "queued"
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/retry")
|
||||
assert r.status_code == 409 and r.json()["code"] == "catalog_draft_busy"
|
||||
|
||||
|
||||
async def test_retry_with_an_unavailable_model_is_409(client, monkeypatch):
|
||||
sid = await _source()
|
||||
monkeypatch.setattr(catalog_jobs.manager, "start", lambda job_id: None)
|
||||
use(monkeypatch, FakeModel(REPLIES))
|
||||
job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"], "use_model": True})).json()["job"]
|
||||
await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/cancel")
|
||||
|
||||
async def unavailable(session):
|
||||
raise catalog.CatalogModelUnavailable("当前只有演示用的模型接入，无法用于起草数据目录")
|
||||
|
||||
monkeypatch.setattr(catalog, "resolve_draft_model", unavailable)
|
||||
r = await client.post(f"{_base(sid)}/draft-jobs/{job['id']}/retry")
|
||||
assert r.status_code == 409 and r.json()["code"] == "catalog_draft_model_unavailable"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 目录清单上的起草状态
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_the_list_carries_last_draft_per_table(client, monkeypatch):
|
||||
sid = await _source()
|
||||
use(monkeypatch, FakeModel(REPLIES, fail={"visits"}))
|
||||
monkeypatch.setattr(catalog, "MODEL_BATCH_TABLES", 1)
|
||||
first = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["visits", "channels"],
|
||||
"use_model": True})).json()["job"]
|
||||
await finished(first["id"])
|
||||
second = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["channels"]})).json()["job"]
|
||||
await finished(second["id"])
|
||||
body = (await client.get(_base(sid))).json()
|
||||
rows = {r["table_name"]: r for r in body["tables"]}
|
||||
assert rows["gates"]["last_draft"] is None
|
||||
visits = rows["visits"]["last_draft"]
|
||||
assert visits["status"] == "failed" and visits["job_id"] == first["id"] and visits["finished_at"]
|
||||
assert "网关超时" in visits["model_error"] and visits["error"] is None
|
||||
assert rows["channels"]["last_draft"]["job_id"] == second["id"], "取最近一次任务里的那一项"
|
||||
assert body["guidance"] is None and body["active_job"] is None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# 起草要求
|
||||
# --------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def test_guidance_is_saved_trimmed_cleared_and_limited(client):
|
||||
sid = await _source()
|
||||
r = await client.put(f"{_base(sid)}/guidance", json={"guidance": " 票务库，金额按元。\n状态 9 是作废 "})
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json() == {"guidance": "票务库，金额按元。\n状态 9 是作废"}
|
||||
assert (await client.get(_base(sid))).json()["guidance"] == "票务库，金额按元。\n状态 9 是作废"
|
||||
too_long = await client.put(f"{_base(sid)}/guidance", json={"guidance": "字" * 2001})
|
||||
assert too_long.status_code == 422 and "2000" in too_long.json()["detail"]
|
||||
assert (await client.get(_base(sid))).json()["guidance"] == "票务库，金额按元。\n状态 9 是作废"
|
||||
assert (await client.put(f"{_base(sid)}/guidance", json={"guidance": "字" * 2000})).status_code == 200
|
||||
cleared = await client.put(f"{_base(sid)}/guidance", json={"guidance": " "})
|
||||
assert cleared.json() == {"guidance": None}
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(DataSource, sid)
|
||||
assert CATALOG_DRAFT_GUIDANCE_OPTION not in (row.options or {})
|
||||
assert (await client.put(f"{_base('missing-source')}/guidance", json={"guidance": "x"})).status_code == 404
|
||||
|
||||
|
||||
async def test_a_table_named_guidance_can_still_be_saved(client):
|
||||
"""起草要求的路径和「整份提交一张表」撞名：表名恰好叫 guidance 时，{notes, if_version} 照旧是那张表的提交，
|
||||
不会被当成清除起草要求；两样都没写的 422。"""
|
||||
tables = _tables()
|
||||
tables["guidance"] = {"qualified": "guidance", "columns": [_col("id"), _col("body", "TEXT")],
|
||||
"primary_key": ["id"], "comment": None, "is_view": False, "foreign_keys": []}
|
||||
sid = await _source(tables)
|
||||
await client.put(f"{_base(sid)}/guidance", json={"guidance": "金额按元"})
|
||||
r = await client.put(f"{_base(sid)}/guidance", json={"notes": {"label": {"value": "操作指引"}}, "if_version": 0},
|
||||
headers=ACTOR)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["table_name"] == "guidance" and r.json()["notes"]["label"]["value"] == "操作指引"
|
||||
assert (await client.get(_base(sid))).json()["guidance"] == "金额按元"
|
||||
assert (await client.put(f"{_base(sid)}/guidance", json={})).status_code == 422
|
||||
|
||||
|
||||
async def test_guidance_of_an_uploaded_source_lives_on_its_own_row(client):
|
||||
sid = await _source(origin="upload")
|
||||
r = await client.put(f"{_base(sid)}/guidance", json={"guidance": "按期上传的客流日报"})
|
||||
assert r.status_code == 200, r.text
|
||||
async with SessionLocal() as session:
|
||||
row = await session.get(DataSource, sid)
|
||||
assert row.options[CATALOG_DRAFT_GUIDANCE_OPTION] == "按期上传的客流日报"
|
||||
|
||||
|
||||
async def test_the_datasource_form_neither_shows_nor_drops_the_guidance(client):
|
||||
sid = await _source(options={"schema": "main"})
|
||||
await client.put(f"{_base(sid)}/guidance", json={"guidance": "金额按元"})
|
||||
shown = (await client.get("/api/datasources")).json()
|
||||
mine = next(s for s in shown if s["id"] == sid)
|
||||
assert CATALOG_DRAFT_GUIDANCE_OPTION not in mine["options"] and mine["options"]["schema"] == "main"
|
||||
# 表单整份回写 options（不带起草要求，或者带着一份别的）：起草要求留着，只认 PUT /catalog/guidance
|
||||
r = await client.patch(f"/api/datasources/{sid}", json={"options": {"schema": "other",
|
||||
CATALOG_DRAFT_GUIDANCE_OPTION: "篡改"}})
|
||||
assert r.status_code == 200, r.text
|
||||
assert CATALOG_DRAFT_GUIDANCE_OPTION not in r.json()["options"]
|
||||
assert (await client.get(_base(sid))).json()["guidance"] == "金额按元"
|
||||
async with SessionLocal() as session:
|
||||
assert (await session.get(DataSource, sid)).options == {"schema": "other",
|
||||
CATALOG_DRAFT_GUIDANCE_OPTION: "金额按元"}
|
||||
|
||||
|
||||
def test_the_guidance_never_reaches_the_driver():
|
||||
src = SimpleNamespace(kind="mysql", host="db.example", port=3306, database="shop", username="reader",
|
||||
password=None, options={"charset": "utf8mb4", CATALOG_DRAFT_GUIDANCE_OPTION: "金额按元"})
|
||||
url = build_url(src)
|
||||
assert "charset=utf8mb4" in url and CATALOG_DRAFT_GUIDANCE_OPTION not in url
|
||||
|
||||
|
||||
async def test_the_sync_draft_uses_the_source_guidance(client, monkeypatch):
|
||||
sid = await _source(description="虚构的景区票务库")
|
||||
model = FakeModel(REPLIES)
|
||||
use(monkeypatch, model)
|
||||
await client.put(f"{_base(sid)}/guidance", json={"guidance": "渠道表的名称是对外展示名"})
|
||||
r = await client.post(f"{_base(sid)}/draft", json={"tables": ["channels"], "use_model": True})
|
||||
assert r.status_code == 200, r.text
|
||||
assert "数据源说明：虚构的景区票务库\n起草要求：渠道表的名称是对外展示名" in model.calls[0]
|
||||
assert set(r.json()) == {"tables", "total", "model_used", "model", "model_error"}, "旧的同步接口返回形状不变"
|
||||
|
||||
|
||||
async def test_jobs_go_away_with_their_source(client):
|
||||
sid = await _source()
|
||||
job = (await client.post(f"{_base(sid)}/draft-jobs", json={"tables": ["gates"]})).json()["job"]
|
||||
await finished(job["id"])
|
||||
await drop_source(sid)
|
||||
_CREATED.remove(sid)
|
||||
async with SessionLocal() as session:
|
||||
assert (await session.execute(select(CatalogDraftJob).where(CatalogDraftJob.id == job["id"]))).first() is None
|
||||
assert (await session.execute(select(CatalogDraftTask).where(CatalogDraftTask.job_id == job["id"]))).first() \
|
||||
is None
|
||||
Reference in New Issue
Block a user