Commit 62f0ce1a authored by Data Governance Dev's avatar Data Governance Dev

feat(web3): P4 任务字段+校验规则持久化

**后端**
- 新增 models/field.py + models/rule.py(两张 ORM:field/task_id FK + rule/field_id FK + ON DELETE CASCADE)
- Field.__table_args__ 加 UNIQUE(task_id, field_key)(schema.sql 第 59 行的事实地约束)
- models/__init__.py 注册新模型;app.py 在 init_db 前 import models(否则 Base.metadata 不知道有这两张表)
- routers/tasks.py:POST/PUT/copy 接 fields+rules;整存整取(先删后写,DB CASCADE 带走 rule);GET 列表给 fields/rules 计数,详情带 field_list;pydantic Field 改 import as PydField 避与 SQLAlchemy Field 同名撞车;port 出参强转 str 防 conn_json 里 int 致 500
- db/database.py:新增 _migrate_field_unique()(仿 _migrate_connection_preset_unique 套路:检测 UNIQUE 缺失则 drop+recreate),init_db 里调用

**前端**
- TaskConfigView.openEdit:拆「同步设 row + 异步 getTask」两步,确保 TaskModal 拿到带 field_list 的详情(列表接口故意不带 field_list 省带宽)
- TaskModal:watch(open) 编辑场景回填 form + 抽 _fillFieldListFromTask;新加 watch(field_list) 处理 getTask 异步时序;loadColumns 重选表时按 field_key 合并 rules 不覆盖
- TaskModal:加 connVerified 标志,testConn 成功才置 true;loadTables/loadColumns 未验证直接 return(编辑场景 form.schema 从 task 回填会触发 watch(form.schema) → loadTables → 空密码连 MySQL 报 1045)

**接口契约**
- POST/PUT /api/tasks body 加 fields: [{key, show_default, rules: [{desc, regex}]}]
- GET /api/tasks 返回 fields/rules 计数;GET /api/tasks/{id} 多返 field_list
- POST /api/tasks/{id}/copy 复制时连同 fields+rules 一起克隆,名字加 _副本 自动避重
- DELETE 走 DB 层 CASCADE(field/rule 一并清)

**端到端**
- POST 3 fields + 3 rules → 列表/详情计数与字段一致
- PUT 整体替换(删 1 加 1 改 1 全部生效)
- POST /copy → field_list 全克隆
- DELETE → SQLite 直查 0 孤儿 field/rule 记录

踩坑(详见 work-logs/2026-08-20.md 主题 28 + 微调 1/2/3)
- pydantic Field 被 SQLAlchemy Field 同名遮蔽(import 时必须 as 别名)
- Base.metadata.create_all 不会改已存在表(UNIQUE 靠 ORM 声明 + 启动时迁移脚本兜底)
- 验 SQLite UNIQUE 不能只看 sqlite_master type='index',UniqueConstraint 在表内 CREATE TABLE 子句里,要 INSERT 重复值测
- 编辑场景的 watch(form.schema) 没拦住自动连数据库(connVerified 守卫)
parent 4aed2229
...@@ -31,6 +31,7 @@ logger.info(f"工作目录: {PROJECT_ROOT}") ...@@ -31,6 +31,7 @@ logger.info(f"工作目录: {PROJECT_ROOT}")
from web3.backend.db.database import init_db, DB_PATH # noqa: E402 from web3.backend.db.database import init_db, DB_PATH # noqa: E402
from web3.backend import models # noqa: F401, E402 (必须 import 所有 ORM 才能被 Base.metadata 知道)
init_db() init_db()
logger.info(f"DB: {DB_PATH} (存在={DB_PATH.exists()})") logger.info(f"DB: {DB_PATH} (存在={DB_PATH.exists()})")
......
...@@ -86,6 +86,12 @@ def init_db() -> None: ...@@ -86,6 +86,12 @@ def init_db() -> None:
# 然后通过 Base.metadata.create_all 增量建出来时只采纳了 ORM 列,丢了 SQL 里的 UNIQUE。 # 然后通过 Base.metadata.create_all 增量建出来时只采纳了 ORM 列,丢了 SQL 里的 UNIQUE。
# 这里检测后重建(connection_preset 里的数据可从 web2 yaml 重新灌入,不影响业务表)。 # 这里检测后重建(connection_preset 里的数据可从 web2 yaml 重新灌入,不影响业务表)。
_migrate_connection_preset_unique() _migrate_connection_preset_unique()
# 一次性迁移:补齐 field 的 UNIQUE(task_id, field_key) 约束
# 历史背景:field 表当初是用 schema.sql 片段手动建出来的(schema.sql 是事实来源),
# 但 ORM 里没声明 UniqueConstraint,Base.metadata.create_all 不会给已存在的表加 UNIQUE。
# 解决:drop field(rule 跟着 ON DELETE CASCADE 一起清)→ create_all 重建带 UNIQUE 的版本。
# 重建后字段/规则数据会丢,需要用户重新保存任务。
_migrate_field_unique()
logger.info(f"[init_db] DB 已存在,已执行增量建表检查:{DB_PATH}") logger.info(f"[init_db] DB 已存在,已执行增量建表检查:{DB_PATH}")
# 灌种子(每个 seed 函数内部判重,可重复调用) # 灌种子(每个 seed 函数内部判重,可重复调用)
seed.seed_task_groups() seed.seed_task_groups()
...@@ -118,4 +124,47 @@ def _migrate_connection_preset_unique() -> None: ...@@ -118,4 +124,47 @@ def _migrate_connection_preset_unique() -> None:
Base.metadata.create_all(engine, tables=[ Base.metadata.create_all(engine, tables=[
t for t in Base.metadata.sorted_tables if t.name == "connection_preset" t for t in Base.metadata.sorted_tables if t.name == "connection_preset"
]) ])
def _migrate_field_unique() -> None:
"""检查 field 表是否已有 UNIQUE(task_id, field_key) 约束,没有则重建 field+rule。
DDL 见 schema.sql 第 51-60 行:
CREATE TABLE field (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task_id INTEGER NOT NULL REFERENCES task(id) ON DELETE CASCADE,
field_key TEXT NOT NULL,
field_cn TEXT,
data_type TEXT,
show_default INTEGER NOT NULL DEFAULT 1,
ord INTEGER NOT NULL DEFAULT 0,
UNIQUE(task_id, field_key)
);
CREATE INDEX idx_field_task ON field(task_id);
历史背景:field 表当初是用 schema.sql 片段手动建出来的(schema.sql 是事实来源),
ORM 里没声明 UniqueConstraint,Base.metadata.create_all 不会给已存在的表加 UNIQUE。
重建策略:drop field(CASCADE 带走 rule)→ Base.metadata.create_all 重建带 UNIQUE 的版本。
重建后字段/规则数据会丢,需要用户重新保存任务。
"""
with engine.begin() as conn:
# 1. 查 field 表是否存在
row = conn.exec_driver_sql(
"SELECT name FROM sqlite_master WHERE type='table' AND name='field'"
).fetchone()
if not row:
return # 表都不在,Base.metadata.create_all 已经建了带 UNIQUE 的版本,无需处理
# 2. 查 UNIQUE 索引是否存在
row = conn.exec_driver_sql(
"SELECT name FROM sqlite_master WHERE type='index' "
"AND tbl_name='field' AND name='uq_field_task_key'"
).fetchone()
if row:
return # 已经有约束了
# 3. 重建表(rule 跟着 ON DELETE CASCADE 一起清;field/rule 数据会丢,需要用户重新保存任务)
logger.warning("[migrate] field 缺 UNIQUE(task_id, field_key) 约束,删除并重建(连带 rule)")
conn.exec_driver_sql("DROP TABLE field")
Base.metadata.create_all(engine, tables=[
t for t in Base.metadata.sorted_tables if t.name in ("field", "rule")
])
logger.info("[migrate] connection_preset 已用带 UNIQUE 约束的版本重建") logger.info("[migrate] connection_preset 已用带 UNIQUE 约束的版本重建")
\ No newline at end of file
"""web3 后端 ORM 模型子包 """web3 后端 ORM 模型子包
每个文件对应 schema.sql 里的一张表。 每个文件对应 schema.sql 里的一张表。
注意:app.py 启动时会触发 init_db(),再调 Base.metadata.create_all() 增量建表。
要求所有 ORM 类必须在本包 __init__ 时被 import 进 Base.metadata,否则建表会被静默跳过。
""" """
from .connection_preset import ConnectionPreset from .connection_preset import ConnectionPreset
from .field import Field
from .rule import Rule
from .task import Task from .task import Task
from .task_group import TaskGroup from .task_group import TaskGroup
__all__ = ["ConnectionPreset", "Task", "TaskGroup"] __all__ = ["ConnectionPreset", "Field", "Rule", "Task", "TaskGroup"]
\ No newline at end of file \ No newline at end of file
"""field ORM 模型
对应 schema.sql 里的:
CREATE TABLE field (
id, task_id, field_key, show_default, ord
)
设计:
- 每个字段属于一个 task(task_id FK + ON DELETE CASCADE)
- 字段级数据字典(cn/data_type)**不持久化**,编辑时按 task_id+source_table 重新从数据源拉
(plan §4 决策:「数据字典不用保存,之后编辑的时候重新获取」)
- rules 关系:一对多挂 Rule(cascade delete-orphan,删 field 自动删 rule)
"""
from __future__ import annotations
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint
from sqlalchemy.orm import relationship
from web3.backend.db.database import Base
class Field(Base):
__tablename__ = "field"
__table_args__ = (
# 同一 task 内 field_key 不可重(schema.sql 设计;ORM 必须声明才能被 create_all 落地)
UniqueConstraint("task_id", "field_key", name="uq_field_task_key"),
)
id = Column(Integer, primary_key=True, autoincrement=True)
task_id = Column(Integer, ForeignKey("task.id", ondelete="CASCADE"), nullable=False)
field_key = Column(String, nullable=False) # 英文字段名(与数据源对齐)
show_default = Column(Integer, nullable=False, default=1) # 0/1:结果表默认展示
ord = Column(Integer, nullable=False, default=0)
rules = relationship(
"Rule",
cascade="all, delete-orphan",
order_by="Rule.ord",
passive_deletes=True,
)
def to_dict(self, *, rules: list | None = None) -> dict:
return {
"id": self.id,
"field_key": self.field_key,
"show_default": bool(self.show_default),
"ord": self.ord,
"rules": rules if rules is not None else [r.to_dict() for r in self.rules],
}
def __repr__(self) -> str:
return f"<Field id={self.id} task_id={self.task_id} field_key={self.field_key!r}>"
\ No newline at end of file
"""rule ORM 模型
对应 schema.sql 里的:
CREATE TABLE rule (
id, field_id, desc, regex, ord
)
设计:
- 每条规则属于一个 field(field_id FK + ON DELETE CASCADE)
- desc:自然语言规则说明(如「必须是 18 位中国大陆身份证号」)
- regex:Python 正则表达式(可空——「非空」这种规则 regex 留空 + 检测时由后端补 `^.+$`)
- ord:规则在 UI 上的显示顺序(保存时按数组顺序写入)
"""
from __future__ import annotations
from sqlalchemy import Column, Integer, Text, ForeignKey
from web3.backend.db.database import Base
class Rule(Base):
__tablename__ = "rule"
id = Column(Integer, primary_key=True, autoincrement=True)
field_id = Column(Integer, ForeignKey("field.id", ondelete="CASCADE"), nullable=False)
desc = Column(Text, nullable=False) # 规则说明(自然语言)
regex = Column(Text, nullable=True) # 正则表达式(可空)
ord = Column(Integer, nullable=False, default=0)
def to_dict(self) -> dict:
return {
"id": self.id,
"desc": self.desc,
"regex": self.regex,
"ord": self.ord,
}
def __repr__(self) -> str:
return f"<Rule id={self.id} field_id={self.field_id} desc={self.desc!r}>"
\ No newline at end of file
...@@ -2,17 +2,19 @@ ...@@ -2,17 +2,19 @@
端点: 端点:
GET /api/tasks 列表(支持 group 分组 + keyword 关键字) GET /api/tasks 列表(支持 group 分组 + keyword 关键字)
POST /api/tasks 新建 POST /api/tasks 新建(头部 + conn_json + fields+rules)
GET /api/tasks/{id} 详情(含 conn_json 解析后的 conn 字段,便于前端回填) GET /api/tasks/{id} 详情(含 conn_json 解析后的 conn 字段 + fields+rules)
PUT /api/tasks/{id} 更新 PUT /api/tasks/{id} 更新(同 POST,整体替换 fields+rules)
DELETE /api/tasks/{id} 删除(级联 field/rule/violation,DB 层 ON DELETE CASCADE) DELETE /api/tasks/{id} 删除(级联 field/rule/violation,DB 层 ON DELETE CASCADE)
POST /api/tasks/{id}/copy 复制(名字加 `_副本`) POST /api/tasks/{id}/copy 复制(名字加 `_副本`,连同 fields+rules 也复制)
设计: 设计:
- 任务表是「头部」,字段/规则放 P4 - 任务表是「头部」+ conn_json;字段/规则放 field/rule 两张表(plan §3 P4)
- name 唯一 → 冲突返回 409 - name 唯一 → 冲突返回 409
- group_id 通过查询参数 group(按名字匹配)传入;前端拿分组名 → 调 POST/PUT 时自动转 group_id - group_id 通过查询参数 group(按名字匹配)传入;前端拿分组名 → 调 POST/PUT 时自动转 group_id
- 数据源路径 / 表名暂存,不参与本阶段校验;前端弹窗仍按 mock 字段展示(由 P5 替换) - 数据字典(field_cn / data_type)**不存**,编辑时按 source_table 重新从数据源拉
(plan §4 决策 + 用户原话)
- fields 整体替换(先删后写):避免「删了哪条、加了哪条」diff 逻辑;任务级 field 数一般在 30 以内,开销可接受
""" """
from __future__ import annotations from __future__ import annotations
...@@ -20,11 +22,13 @@ import json ...@@ -20,11 +22,13 @@ import json
from typing import Optional from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field from pydantic import BaseModel, Field as PydField
from sqlalchemy.exc import IntegrityError from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from web3.backend.db.database import get_session from web3.backend.db.database import get_session
from web3.backend.models.field import Field
from web3.backend.models.rule import Rule
from web3.backend.models.task import Task from web3.backend.models.task import Task
from web3.backend.models.task_group import TaskGroup from web3.backend.models.task_group import TaskGroup
...@@ -33,19 +37,30 @@ router = APIRouter(prefix="/tasks", tags=["tasks"]) ...@@ -33,19 +37,30 @@ router = APIRouter(prefix="/tasks", tags=["tasks"])
# ── Pydantic schemas ────────────────────────────────────────────── # ── Pydantic schemas ──────────────────────────────────────────────
class RulePayload(BaseModel):
desc: str = PydField(..., min_length=1, description="规则说明(自然语言)")
regex: Optional[str] = PydField(None, description="Python 正则表达式(可空)")
class FieldPayload(BaseModel):
key: str = PydField(..., min_length=1, description="英文字段名(与数据源对齐)")
show_default: bool = PydField(True, description="结果表是否默认展示")
rules: list[RulePayload] = PydField(default_factory=list, description="字段下的规则列表")
class TaskBase(BaseModel): class TaskBase(BaseModel):
name: str = Field(..., min_length=1, max_length=100) name: str = PydField(..., min_length=1, max_length=100)
group: str = Field(..., description="分组名(前端传名,服务端查 id)") group: str = PydField(..., description="分组名(前端传名,服务端查 id)")
db_type: str = Field("MySQL", description="UI 冗余,不参与检测") db_type: str = PydField("MySQL", description="UI 冗余,不参与检测")
data_source_path: Optional[str] = Field(None, description="如 sources/customer_db/customer.sqlite") data_source_path: Optional[str] = PydField(None, description="如 sources/customer_db/customer.sqlite")
source_table: Optional[str] = Field(None, description="如 t_user_info") source_table: Optional[str] = PydField(None, description="如 t_user_info")
status: str = Field("启用", description="启用 / 停用") status: str = PydField("启用", description="启用 / 停用")
description: Optional[str] = None description: Optional[str] = None
conn_json: Optional[str] = Field(None, description="UI 数据连接表单的 JSON 字符串") conn_json: Optional[str] = PydField(None, description="UI 数据连接表单的 JSON 字符串")
class TaskCreate(TaskBase): class TaskCreate(TaskBase):
pass fields: list[FieldPayload] = PydField(default_factory=list, description="任务的字段+规则(整体写入)")
class TaskUpdate(BaseModel): class TaskUpdate(BaseModel):
...@@ -57,6 +72,22 @@ class TaskUpdate(BaseModel): ...@@ -57,6 +72,22 @@ class TaskUpdate(BaseModel):
status: Optional[str] = None status: Optional[str] = None
description: Optional[str] = None description: Optional[str] = None
conn_json: Optional[str] = None conn_json: Optional[str] = None
fields: Optional[list[FieldPayload]] = PydField(None, description="提供则整体替换;不传则保留旧 fields")
class FieldOut(BaseModel):
id: int
field_key: str
show_default: bool
ord: int
rules: list["RuleOut"] = []
class RuleOut(BaseModel):
id: int
desc: str
regex: Optional[str]
ord: int
class TaskOut(BaseModel): class TaskOut(BaseModel):
...@@ -77,8 +108,9 @@ class TaskOut(BaseModel): ...@@ -77,8 +108,9 @@ class TaskOut(BaseModel):
source_table: Optional[str] source_table: Optional[str]
status: str status: str
description: Optional[str] description: Optional[str]
fields: int fields: int # 字段数(列表视图用)
rules: int rules: int # 规则总数(列表视图用)
field_list: list[FieldOut] = [] # 详情用(含 rules);列表视图为空(节省带宽)
checked_at: Optional[str] checked_at: Optional[str]
created_at: str created_at: str
updated_at: str updated_at: str
...@@ -103,9 +135,93 @@ def _parse_conn_json(conn_json: Optional[str]) -> dict: ...@@ -103,9 +135,93 @@ def _parse_conn_json(conn_json: Optional[str]) -> dict:
return {} return {}
def _row_to_out(t: Task) -> TaskOut: def _replace_fields(db: Session, task: Task, fields_payload: list[FieldPayload]) -> None:
"""整体替换 task 下的 fields + rules(先清空再插入)。
为什么整体替换而不是 diff:
- 字段/规则条数小(一般 30 字段内、每字段 1-3 条规则),开销可接受
- 避免「删了哪条、加了哪条、改了哪条」三向 diff 逻辑,事务里一行 db.execute 就够
- 顺序由前端数组顺序决定 → ord 字段直接用 enumerate 写
"""
# 1) 清空旧 field(DB 层 ON DELETE CASCADE 会把 rules 也带走)
db.query(Field).filter(Field.task_id == task.id).delete(synchronize_session=False)
db.flush()
# 2) 写入新 field + rule
for f_idx, f in enumerate(fields_payload):
field = Field(
task_id=task.id,
field_key=f.key,
show_default=1 if f.show_default else 0,
ord=f_idx,
)
db.add(field)
db.flush() # 拿到 field.id
for r_idx, r in enumerate(f.rules):
rule = Rule(
field_id=field.id,
desc=r.desc.strip(),
regex=(r.regex or "").strip() or None,
ord=r_idx,
)
db.add(rule)
def _copy_fields(db: Session, src_task: Task, dst_task: Task) -> None:
"""复制 src_task 的 fields+rules 到 dst_task(用于 copy 端点)。"""
src_fields = db.query(Field).filter(Field.task_id == src_task.id).order_by(Field.ord).all()
for f_idx, src_f in enumerate(src_fields):
new_f = Field(
task_id=dst_task.id,
field_key=src_f.field_key,
show_default=src_f.show_default,
ord=f_idx,
)
db.add(new_f)
db.flush()
for r_idx, src_r in enumerate(
db.query(Rule).filter(Rule.field_id == src_f.id).order_by(Rule.ord).all()
):
db.add(Rule(
field_id=new_f.id,
desc=src_r.desc,
regex=src_r.regex,
ord=r_idx,
))
def _row_to_out(db: Session, t: Task, *, include_field_list: bool = False) -> TaskOut:
"""t.to_dict() 给基础字段;counts / field_list 都复用当前 session 直接 query。
include_field_list=True 时再查 field+rule 全量(详情用);
列表页不查(避免 N+1)。
"""
d = t.to_dict() d = t.to_dict()
conn = _parse_conn_json(t.conn_json) conn = _parse_conn_json(t.conn_json)
fields_count = (
db.query(Field).filter(Field.task_id == t.id).count() if t.id else 0
)
rules_count = (
db.query(Rule).join(Field, Rule.field_id == Field.id)
.filter(Field.task_id == t.id).count() if t.id else 0
)
field_list: list[FieldOut] = []
if include_field_list:
for f in db.query(Field).filter(Field.task_id == t.id).order_by(Field.ord).all():
rules = [
RuleOut(id=r.id, desc=r.desc, regex=r.regex, ord=r.ord)
for r in db.query(Rule).filter(Rule.field_id == f.id).order_by(Rule.ord).all()
]
field_list.append(FieldOut(
id=f.id,
field_key=f.field_key,
show_default=bool(f.show_default),
ord=f.ord,
rules=rules,
))
return TaskOut( return TaskOut(
id=d["id"], id=d["id"],
name=d["name"], name=d["name"],
...@@ -115,7 +231,7 @@ def _row_to_out(t: Task) -> TaskOut: ...@@ -115,7 +231,7 @@ def _row_to_out(t: Task) -> TaskOut:
db=conn.get("db"), db=conn.get("db"),
conn_name=conn.get("connName"), conn_name=conn.get("connName"),
host=conn.get("host"), host=conn.get("host"),
port=conn.get("port"), port=(str(conn["port"]) if conn.get("port") is not None else None),
user=conn.get("user"), user=conn.get("user"),
jdbc_params=conn.get("jdbcParams"), jdbc_params=conn.get("jdbcParams"),
schema=conn.get("schema"), schema=conn.get("schema"),
...@@ -123,8 +239,9 @@ def _row_to_out(t: Task) -> TaskOut: ...@@ -123,8 +239,9 @@ def _row_to_out(t: Task) -> TaskOut:
source_table=d["source_table"], source_table=d["source_table"],
status=d["status"], status=d["status"],
description=d["description"], description=d["description"],
fields=d["fields"], fields=fields_count,
rules=d["rules"], rules=rules_count,
field_list=field_list,
checked_at=d["checked_at"], checked_at=d["checked_at"],
created_at=d["created_at"], created_at=d["created_at"],
updated_at=d["updated_at"], updated_at=d["updated_at"],
...@@ -149,7 +266,7 @@ def list_tasks( ...@@ -149,7 +266,7 @@ def list_tasks(
like = f"%{keyword}%" like = f"%{keyword}%"
q = q.filter((Task.name.like(like)) | (Task.source_table.like(like))) q = q.filter((Task.name.like(like)) | (Task.source_table.like(like)))
rows = q.order_by(Task.id.desc()).all() rows = q.order_by(Task.id.desc()).all()
return [_row_to_out(t) for t in rows] return [_row_to_out(db, t) for t in rows]
@router.post("", response_model=TaskOut, summary="新建任务") @router.post("", response_model=TaskOut, summary="新建任务")
...@@ -167,29 +284,43 @@ def create_task(payload: TaskCreate, db: Session = Depends(get_session)) -> Task ...@@ -167,29 +284,43 @@ def create_task(payload: TaskCreate, db: Session = Depends(get_session)) -> Task
) )
db.add(t) db.add(t)
try: try:
db.commit() db.flush() # 拿到 t.id
except IntegrityError: except IntegrityError:
db.rollback() db.rollback()
raise HTTPException(status_code=409, detail=f"任务名已存在:{payload.name}") raise HTTPException(status_code=409, detail=f"任务名已存在:{payload.name}")
# 写 fields + rules(一个事务,rollback 则整体回滚)
try:
_replace_fields(db, t, payload.fields)
db.commit()
except IntegrityError:
db.rollback()
raise HTTPException(status_code=409, detail="field_key 重复或约束冲突")
except Exception as e:
db.rollback()
raise HTTPException(status_code=500, detail=f"写 fields/rule 失败:{type(e).__name__}: {e}")
db.refresh(t) db.refresh(t)
return _row_to_out(t) return _row_to_out(db, t, include_field_list=True)
@router.get("/{task_id}", response_model=TaskOut, summary="任务详情") @router.get("/{task_id}", response_model=TaskOut, summary="任务详情(含 fields+rules)")
def get_task(task_id: int, db: Session = Depends(get_session)) -> TaskOut: def get_task(task_id: int, db: Session = Depends(get_session)) -> TaskOut:
t = db.query(Task).filter(Task.id == task_id).first() t = db.query(Task).filter(Task.id == task_id).first()
if not t: if not t:
raise HTTPException(status_code=404, detail=f"任务不存在:{task_id}") raise HTTPException(status_code=404, detail=f"任务不存在:{task_id}")
return _row_to_out(t) return _row_to_out(db, t, include_field_list=True)
@router.put("/{task_id}", response_model=TaskOut, summary="更新任务") @router.put("/{task_id}", response_model=TaskOut, summary="更新任务(可选整体替换 fields+rules)")
def update_task(task_id: int, payload: TaskUpdate, db: Session = Depends(get_session)) -> TaskOut: def update_task(task_id: int, payload: TaskUpdate, db: Session = Depends(get_session)) -> TaskOut:
t = db.query(Task).filter(Task.id == task_id).first() t = db.query(Task).filter(Task.id == task_id).first()
if not t: if not t:
raise HTTPException(status_code=404, detail=f"任务不存在:{task_id}") raise HTTPException(status_code=404, detail=f"任务不存在:{task_id}")
data = payload.model_dump(exclude_unset=True) data = payload.model_dump(exclude_unset=True)
fields_payload = data.pop("fields", None) # fields 单独处理(不直接 setattr)
if "group" in data: if "group" in data:
t.group_id = _resolve_group_id(db, data.pop("group")) t.group_id = _resolve_group_id(db, data.pop("group"))
...@@ -205,13 +336,21 @@ def update_task(task_id: int, payload: TaskUpdate, db: Session = Depends(get_ses ...@@ -205,13 +336,21 @@ def update_task(task_id: int, payload: TaskUpdate, db: Session = Depends(get_ses
for k, v in data.items(): for k, v in data.items():
setattr(t, k, v) setattr(t, k, v)
# fields 整体替换(如果 payload 里给了)
if fields_payload is not None:
try:
_replace_fields(db, t, [FieldPayload(**f) for f in fields_payload])
except Exception as e:
db.rollback()
raise HTTPException(status_code=500, detail=f"替换 fields/rule 失败:{type(e).__name__}: {e}")
try: try:
db.commit() db.commit()
except IntegrityError: except IntegrityError:
db.rollback() db.rollback()
raise HTTPException(status_code=409, detail=f"任务名已存在:{payload.name}") raise HTTPException(status_code=409, detail=f"任务名已存在:{payload.name}")
db.refresh(t) db.refresh(t)
return _row_to_out(t) return _row_to_out(db, t, include_field_list=True)
@router.delete("/{task_id}", status_code=204, summary="删除任务") @router.delete("/{task_id}", status_code=204, summary="删除任务")
...@@ -223,7 +362,7 @@ def delete_task(task_id: int, db: Session = Depends(get_session)): ...@@ -223,7 +362,7 @@ def delete_task(task_id: int, db: Session = Depends(get_session)):
db.commit() db.commit()
@router.post("/{task_id}/copy", response_model=TaskOut, summary="复制任务(名字加 _副本)") @router.post("/{task_id}/copy", response_model=TaskOut, summary="复制任务(名字加 _副本,连同 fields+rules)")
def copy_task(task_id: int, db: Session = Depends(get_session)) -> TaskOut: def copy_task(task_id: int, db: Session = Depends(get_session)) -> TaskOut:
t = db.query(Task).filter(Task.id == task_id).first() t = db.query(Task).filter(Task.id == task_id).first()
if not t: if not t:
...@@ -242,12 +381,17 @@ def copy_task(task_id: int, db: Session = Depends(get_session)) -> TaskOut: ...@@ -242,12 +381,17 @@ def copy_task(task_id: int, db: Session = Depends(get_session)) -> TaskOut:
group_id=t.group_id, group_id=t.group_id,
data_source_path=t.data_source_path, data_source_path=t.data_source_path,
source_table=t.source_table, source_table=t.source_table,
status=t.status or "启用", # 沿用源任务 status;P3 阶段 UI 不暴露 status 列 status=t.status or "启用",
description=t.description, description=t.description,
db_type=t.db_type, db_type=t.db_type,
conn_json=t.conn_json, conn_json=t.conn_json,
) )
db.add(clone) db.add(clone)
db.flush()
# 复制 fields + rules
_copy_fields(db, t, clone)
db.commit() db.commit()
db.refresh(clone) db.refresh(clone)
return _row_to_out(clone) return _row_to_out(db, clone, include_field_list=True)
\ No newline at end of file \ No newline at end of file
...@@ -322,6 +322,11 @@ const form = reactive({ ...@@ -322,6 +322,11 @@ const form = reactive({
const testing = ref(false) const testing = ref(false)
const connResult = ref({ ok: null, message: '' }) const connResult = ref({ ok: null, message: '' })
// 连接是否验证过:只有 testConn 成功后才能自动拉 schema/表/字段
// 编辑场景 form.schema 会被从 task 回填,会触发 watch(form.schema) → loadTables → 报错
// 用 connVerified 拦住:未验证过就直接 return,不去连数据库
const connVerified = ref(false)
async function testConn() { async function testConn() {
// 前端必填校验 // 前端必填校验
if (!form.host || !form.port || !form.db || !form.user) { if (!form.host || !form.port || !form.db || !form.user) {
...@@ -334,15 +339,18 @@ async function testConn() { ...@@ -334,15 +339,18 @@ async function testConn() {
const r = await testConnection(form) const r = await testConnection(form)
connResult.value = { ok: r.ok, message: r.message } connResult.value = { ok: r.ok, message: r.message }
if (r.ok) { if (r.ok) {
connVerified.value = true // 标记「已验证」,后续 schema/表/字段自动加载才允许
ElMessage.success(r.message) ElMessage.success(r.message)
// 成功 → 自动拉 schema → 表 // 成功 → 自动拉 schema → 表
await loadSchemas() await loadSchemas()
} else { } else {
connVerified.value = false
ElMessage.error(r.message) ElMessage.error(r.message)
schemaOptions.value = [] schemaOptions.value = []
tableOptions.value = [] tableOptions.value = []
} }
} catch (e) { } catch (e) {
connVerified.value = false
connResult.value = { ok: false, message: e.message } connResult.value = { ok: false, message: e.message }
ElMessage.error(`测试失败:${e.message}`) ElMessage.error(`测试失败:${e.message}`)
schemaOptions.value = [] schemaOptions.value = []
...@@ -382,7 +390,8 @@ async function loadSchemas() { ...@@ -382,7 +390,8 @@ async function loadSchemas() {
} }
async function loadTables() { async function loadTables() {
if (!form.schema) { if (!form.schema || !connVerified.value) {
// 未验证连接 → 不去连数据库(编辑场景 form.schema 从 task 回填时触发,这里直接拦住)
tableOptions.value = [] tableOptions.value = []
return return
} }
...@@ -413,11 +422,15 @@ async function loadTables() { ...@@ -413,11 +422,15 @@ async function loadTables() {
// 选完表后 → 拉该表的全部字段(数据字典),填进校验规则 Tab 的字段列表 // 选完表后 → 拉该表的全部字段(数据字典),填进校验规则 Tab 的字段列表
async function loadColumns() { async function loadColumns() {
if (!form.table || !form.schema) { if (!form.table || !form.schema || !connVerified.value) {
fieldList.value = [] // 未验证连接 / 没选表 → 不拉字段
// 重要:编辑场景下不要清空 fieldList —— _fillFieldListFromTask 已经把后端存的 rules 回填进去了
return return
} }
loadingColumns.value = true loadingColumns.value = true
// 备份旧 rules(按 field_key 索引),新字段列表合并时按 key 找回——避免「重选表」把已配的规则清掉
const rulesByKey = new Map(fieldList.value.map((f) => [f.key, f.rules || []]))
const showDefaultByKey = new Map(fieldList.value.map((f) => [f.key, f.showDefault]))
try { try {
const r = await listColumns(form, form.table) const r = await listColumns(form, form.table)
if (r.ok) { if (r.ok) {
...@@ -427,8 +440,10 @@ async function loadColumns() { ...@@ -427,8 +440,10 @@ async function loadColumns() {
cn: c.column_comment || '', // 字段注释(不可修改;无注释显示 —) cn: c.column_comment || '', // 字段注释(不可修改;无注释显示 —)
type: c.column_type || c.data_type || '', // 数据类型显示(VARCHAR(50) 等) type: c.column_type || c.data_type || '', // 数据类型显示(VARCHAR(50) 等)
isNullable: c.is_nullable, // Y/N(后续 P4 必填校验可能用) isNullable: c.is_nullable, // Y/N(后续 P4 必填校验可能用)
rules: [], // P4 接 rule 表后再填 rules: rulesByKey.get(c.column_name) || [], // 合并旧 rules;新字段默认空
showDefault: true, showDefault: showDefaultByKey.has(c.column_name)
? showDefaultByKey.get(c.column_name)
: true, // 旧值优先;新字段默认 true
})) }))
if (fieldList.value.length === 0) { if (fieldList.value.length === 0) {
ElMessage.warning(`表「${form.table}」下未找到任何字段`) ElMessage.warning(`表「${form.table}」下未找到任何字段`)
...@@ -466,6 +481,7 @@ watch(() => form.dbType, async (newVal) => { ...@@ -466,6 +481,7 @@ watch(() => form.dbType, async (newVal) => {
form.schema = '' form.schema = ''
form.table = '' form.table = ''
connResult.value = { ok: null, message: '' } connResult.value = { ok: null, message: '' }
connVerified.value = false // 切 dbType → 旧验证失效,等用户重新测试连接
selectedPresetId.value = null // 切 dbType 后清掉选中(当前 dbType 的预设列表会刷新) selectedPresetId.value = null // 切 dbType 后清掉选中(当前 dbType 的预设列表会刷新)
await loadPresets(newVal) await loadPresets(newVal)
}) })
...@@ -474,11 +490,33 @@ watch(() => form.dbType, async (newVal) => { ...@@ -474,11 +490,33 @@ watch(() => form.dbType, async (newVal) => {
// 之前用 fieldsByType mock + form.group 反推,2026-08-20 起改为按所选表真实字段填充 // 之前用 fieldsByType mock + form.group 反推,2026-08-20 起改为按所选表真实字段填充
const fieldList = ref([]) const fieldList = ref([])
// 把后端 field_list 回填到字段表(只在编辑场景用)
// 注意:openEdit 是「同步设 row(field_list=[])+ 异步 getTask」两步,必须等 getTask 拿到详情再回填,
// 不能只在 watch(open) 里写一次,否则看到的还是 list 接口的空 field_list。
function _fillFieldListFromTask() {
if (!props.task) return
const fl = props.task.field_list
if (!Array.isArray(fl) || fl.length === 0) return
fieldList.value = fl.map((f) => ({
key: f.field_key,
en: f.field_key,
cn: '',
type: '',
isNullable: '',
rules: f.rules || [],
showDefault: f.show_default !== false,
}))
}
watch(() => props.task?.field_list, () => {
if (props.open) _fillFieldListFromTask()
})
watch(() => props.open, async (v) => { watch(() => props.open, async (v) => {
if (v) { if (v) {
activeTab.value = 'basic' activeTab.value = 'basic'
testing.value = false testing.value = false
connResult.value = { ok: null, message: '' } connResult.value = { ok: null, message: '' }
connVerified.value = false // 打开弹窗时清掉:编辑场景密码不回填,旧验证无效
schemaOptions.value = [] schemaOptions.value = []
tableOptions.value = [] tableOptions.value = []
selectedPresetId.value = null selectedPresetId.value = null
...@@ -502,6 +540,10 @@ watch(() => props.open, async (v) => { ...@@ -502,6 +540,10 @@ watch(() => props.open, async (v) => {
// 数据表:后端存的是「t_user_info」纯名,前端直接展示 // 数据表:后端存的是「t_user_info」纯名,前端直接展示
form.table = props.task.source_table || '' form.table = props.task.source_table || ''
// 注:编辑场景不自动拉 schema/表 —— 等用户点「测试连接」再拉 // 注:编辑场景不自动拉 schema/表 —— 等用户点「测试连接」再拉
// 编辑场景:把详情里的 field_list(带 rules)回填到字段表,省得用户重新走一遍
// cn/data_type 是「编辑时重新取」的——不回填,留空让用户重新测试连接后从 listColumns 补
// field_list 可能还没到(getTask 异步)—— 下面那个 watch 会在拿到后自动补
_fillFieldListFromTask()
} else { } else {
// 新增场景:默认值(不写死假数据) // 新增场景:默认值(不写死假数据)
form.name = '' form.name = ''
...@@ -517,8 +559,8 @@ watch(() => props.open, async (v) => { ...@@ -517,8 +559,8 @@ watch(() => props.open, async (v) => {
form.jdbcParams = '' form.jdbcParams = ''
form.schema = '' form.schema = ''
form.table = '' form.table = ''
fieldList.value = []
} }
fieldList.value = []
} }
}) })
......
...@@ -74,7 +74,7 @@ ...@@ -74,7 +74,7 @@
import { computed, inject, onMounted, ref } from 'vue' import { computed, inject, onMounted, ref } from 'vue'
import { ElMessage, ElMessageBox } from 'element-plus' import { ElMessage, ElMessageBox } from 'element-plus'
import { import {
listTasks, createTask, updateTask, deleteTask, copyTask, listTasks, getTask, createTask, updateTask, deleteTask, copyTask,
} from '@/api/tasks' } from '@/api/tasks'
import TaskModal from '@/components/TaskModal.vue' import TaskModal from '@/components/TaskModal.vue'
...@@ -134,6 +134,14 @@ function openEdit(row) { ...@@ -134,6 +134,14 @@ function openEdit(row) {
modalMode.value = 'edit' modalMode.value = 'edit'
editingTask.value = row editingTask.value = row
modalOpen.value = true modalOpen.value = true
// 编辑要拿详情(field_list + rules),列表接口为了节省带宽不带 field_list
;(async () => {
try {
editingTask.value = await getTask(row.id)
} catch (e) {
ElMessage.error(`加载任务详情失败:${e.message}`)
}
})()
} }
async function onSaved(payload) { async function onSaved(payload) {
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment