Commit 04e0282c authored by Data Governance Dev's avatar Data Governance Dev

feat(web3): P3 任务 CRUD(头部)

后端:
- models/task.py:ORM(name/group_id/data_source_path/source_table/status/description/db_type/conn_json/timestamps)
  - to_dict() 含 group 名字 + fields_count/rules_count(占位 0,P4 改 join)
- routers/tasks.py:6 端点
  - GET /api/tasks?group=&keyword=(keyword 同时匹配 name 和 source_table)
  - POST(name 重名 → 409)、GET /{id}、PUT /{id}、DELETE /{id}(级联走 DB ON DELETE CASCADE)
  - POST /{id}/copy:自动生成不冲突的 xxx_副本 / xxx_副本2 ...
  - group_id 通过 group 名查询
  - conn_json 解析出 db 字段返回
- app.py:注册 tasks 路由

前端:
- src/api/tasks.js:listTasks / getTask / createTask / updateTask / deleteTask / copyTask
- TaskConfigView.vue:
  - 完全去掉 mockData.taskList,onMounted 拉 /api/tasks
  - 新增/编辑/复制/删除全走 API,错误 ElMessage 提示
- TaskModal.vue:onSave 发 {name, group, db_type, data_source_path, source_table, status, description, conn_json}(P3 不写 fields/rules,留 P4)

验收(经 Vite 代理,10 步链路):
  空列表 → 新建 → 重名 409 → 分组筛选 → 关键字筛选 → 复制 → 更新 → 404 → 删除 → 剩 N 条
parent 337a9ad3
......@@ -10,7 +10,6 @@ WEB3_DIR = BACKEND_DIR.parent
PROJECT_ROOT = WEB3_DIR.parent
sys.path.insert(0, str(PROJECT_ROOT))
# Windows GBK console 兼容
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
sys.stderr.reconfigure(encoding="utf-8", errors="replace")
......@@ -31,20 +30,17 @@ logger.info(f"Python: {sys.version.split()[0]}, 平台: {sys.platform}")
logger.info(f"工作目录: {PROJECT_ROOT}")
# ── DB 首次启动初始化 ──
from web3.backend.db.database import init_db, DB_PATH # noqa: E402
init_db()
logger.info(f"DB: {DB_PATH} (存在={DB_PATH.exists()})")
# ── 创建 FastAPI 应用 ──
app = FastAPI(
title="数据治理 Web 工具 · web3 后端",
description="数据质量检测(任务 / 字段 / 规则 / 不合规)。前端在 web3/,由 Vite 5175 提供。",
version="0.1.0",
)
# CORS(开发模式:5175 是 Vite dev server)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
......@@ -54,23 +50,22 @@ app.add_middleware(
)
# ── 注册路由 ──
from web3.backend.routers.task_groups import router as task_groups_router # noqa: E402
from web3.backend.routers.tasks import router as tasks_router # noqa: E402
app.include_router(task_groups_router, prefix="/api")
logger.info("已注册路由:/api/task-groups")
app.include_router(tasks_router, prefix="/api")
logger.info("已注册路由:/api/task-groups, /api/tasks")
# ── 健康检查 ──
@app.get("/api/health")
async def health():
return {"status": "ok", "service": "web3-backend"}
# ── 根路径(开发模式提示) ──
@app.get("/", response_class=HTMLResponse)
async def root():
return HTMLResponse(
"<h1>web3 后端 API 服务</h1>"
"<p>开发模式请用 <code>npm run dev</code> 启动 Vite(端口 5175)。</p>"
"<p>API 文档:<a href='/docs'>/docs</a></p>"
)
\ No newline at end of file
)
"""task ORM 模型
对应 schema.sql 里的:
CREATE TABLE task (
id, name, group_id, data_source_path, source_table,
status, description, db_type, conn_json, created_at, updated_at
)
设计:
- 一个 task = 一个分组 + 一张源表 + 一组字段(field) + 一组规则(rule) + 一次最近检测(validation_run)
- conn_json:UI 上数据连接表单的 JSON 冗余存一份(仅展示,不参与计算)
- 真实检测读本地 SQLite,不读 conn_json
"""
from __future__ import annotations
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from web3.backend.db.database import Base
from web3.backend.models.task_group import TaskGroup
class Task(Base):
__tablename__ = "task"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String, nullable=False, unique=True)
group_id = Column(Integer, ForeignKey("task_group.id"), nullable=False)
data_source_path = Column(Text, nullable=True) # 例如 sources/customer_db/customer.sqlite
source_table = Column(String, nullable=True) # 例如 t_user_info
status = Column(String, nullable=False, default="启用") # 启用 / 停用
description = Column(Text, nullable=True)
db_type = Column(String, nullable=False, default="MySQL") # UI 展示冗余
conn_json = Column(Text, nullable=True) # UI 表单 JSON
created_at = Column(DateTime, nullable=False, server_default=func.current_timestamp())
updated_at = Column(DateTime, nullable=False,
server_default=func.current_timestamp(),
onupdate=func.current_timestamp())
# 关联(不强制 eager load,需要时再 query)
group = relationship("TaskGroup", lazy="joined")
def to_dict(self, *, fields_count: int = 0, rules_count: int = 0) -> dict:
"""序列化为前端友好结构
- fields_count / rules_count 由 router 层统计后传入(P4 之后改 join)
"""
return {
"id": self.id,
"name": self.name,
"group_id": self.group_id,
"group": self.group.name if self.group else None,
"data_source_path": self.data_source_path,
"source_table": self.source_table,
"status": self.status,
"description": self.description,
"db_type": self.db_type,
"conn_json": self.conn_json,
"fields": fields_count,
"rules": rules_count,
"checked_at": None, # P6/P7 检测引擎接上后再回填
"created_at": self.created_at.isoformat() if isinstance(self.created_at, datetime) else self.created_at,
"updated_at": self.updated_at.isoformat() if isinstance(self.updated_at, datetime) else self.updated_at,
}
def __repr__(self) -> str:
return f"<Task id={self.id} name={self.name!r} group_id={self.group_id}>"
\ No newline at end of file
"""任务路由
端点:
GET /api/tasks 列表(支持 group 分组 + keyword 关键字)
POST /api/tasks 新建
GET /api/tasks/{id} 详情(含 conn_json 解析后的 conn 字段,便于前端回填)
PUT /api/tasks/{id} 更新
DELETE /api/tasks/{id} 删除(级联 field/rule/violation,DB 层 ON DELETE CASCADE)
POST /api/tasks/{id}/copy 复制(名字加 `_副本`)
设计:
- 任务表是「头部」,字段/规则放 P4
- name 唯一 → 冲突返回 409
- group_id 通过查询参数 group(按名字匹配)传入;前端拿分组名 → 调 POST/PUT 时自动转 group_id
- 数据源路径 / 表名暂存,不参与本阶段校验;前端弹窗仍按 mock 字段展示(由 P5 替换)
"""
from __future__ import annotations
import json
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from web3.backend.db.database import get_session
from web3.backend.models.task import Task
from web3.backend.models.task_group import TaskGroup
router = APIRouter(prefix="/tasks", tags=["tasks"])
# ── Pydantic schemas ──────────────────────────────────────────────
class TaskBase(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
group: str = Field(..., description="分组名(前端传名,服务端查 id)")
db_type: str = Field("MySQL", description="UI 冗余,不参与检测")
data_source_path: Optional[str] = Field(None, description="如 sources/customer_db/customer.sqlite")
source_table: Optional[str] = Field(None, description="如 t_user_info")
status: str = Field("启用", description="启用 / 停用")
description: Optional[str] = None
conn_json: Optional[str] = Field(None, description="UI 数据连接表单的 JSON 字符串")
class TaskCreate(TaskBase):
pass
class TaskUpdate(BaseModel):
name: Optional[str] = None
group: Optional[str] = None
db_type: Optional[str] = None
data_source_path: Optional[str] = None
source_table: Optional[str] = None
status: Optional[str] = None
description: Optional[str] = None
conn_json: Optional[str] = None
class TaskOut(BaseModel):
id: int
name: str
group_id: int
group: Optional[str]
db_type: str
db: Optional[str] = None # 从 conn_json 解析出来的「数据库名」,给前端 TaskConfigView 用
data_source_path: Optional[str]
source_table: Optional[str]
status: str
description: Optional[str]
fields: int
rules: int
checked_at: Optional[str]
created_at: str
updated_at: str
# ── helpers ───────────────────────────────────────────────────────
def _resolve_group_id(db: Session, group_name: str) -> int:
g = db.query(TaskGroup).filter(TaskGroup.name == group_name).first()
if not g:
raise HTTPException(status_code=400, detail=f"分组不存在:{group_name}")
return g.id
def _row_to_out(t: Task) -> TaskOut:
d = t.to_dict()
# 从 conn_json 里抠一个 db(数据库名)出来给前端展示
db_name = None
if t.conn_json:
try:
db_name = json.loads(t.conn_json).get("db")
except Exception:
pass
return TaskOut(
id=d["id"],
name=d["name"],
group_id=d["group_id"],
group=d["group"],
db_type=d["db_type"],
db=db_name,
data_source_path=d["data_source_path"],
source_table=d["source_table"],
status=d["status"],
description=d["description"],
fields=d["fields"],
rules=d["rules"],
checked_at=d["checked_at"],
created_at=d["created_at"],
updated_at=d["updated_at"],
)
# ── 路由 ──────────────────────────────────────────────────────────
@router.get("", response_model=list[TaskOut], summary="任务列表(支持 group/keyword 筛选)")
def list_tasks(
group: Optional[str] = Query(None, description="按分组名过滤"),
keyword: Optional[str] = Query(None, description="按名称 / 表名模糊匹配"),
db: Session = Depends(get_session),
) -> list[TaskOut]:
q = db.query(Task)
if group:
g = db.query(TaskGroup).filter(TaskGroup.name == group).first()
if not g:
return []
q = q.filter(Task.group_id == g.id)
if keyword:
like = f"%{keyword}%"
q = q.filter((Task.name.like(like)) | (Task.source_table.like(like)))
rows = q.order_by(Task.id.desc()).all()
return [_row_to_out(t) for t in rows]
@router.post("", response_model=TaskOut, summary="新建任务")
def create_task(payload: TaskCreate, db: Session = Depends(get_session)) -> TaskOut:
group_id = _resolve_group_id(db, payload.group)
t = Task(
name=payload.name.strip(),
group_id=group_id,
data_source_path=payload.data_source_path,
source_table=payload.source_table,
status=payload.status or "启用",
description=payload.description,
db_type=payload.db_type or "MySQL",
conn_json=payload.conn_json,
)
db.add(t)
try:
db.commit()
except IntegrityError:
db.rollback()
raise HTTPException(status_code=409, detail=f"任务名已存在:{payload.name}")
db.refresh(t)
return _row_to_out(t)
@router.get("/{task_id}", response_model=TaskOut, summary="任务详情")
def get_task(task_id: int, db: Session = Depends(get_session)) -> TaskOut:
t = db.query(Task).filter(Task.id == task_id).first()
if not t:
raise HTTPException(status_code=404, detail=f"任务不存在:{task_id}")
return _row_to_out(t)
@router.put("/{task_id}", response_model=TaskOut, summary="更新任务")
def update_task(task_id: int, payload: TaskUpdate, db: Session = Depends(get_session)) -> TaskOut:
t = db.query(Task).filter(Task.id == task_id).first()
if not t:
raise HTTPException(status_code=404, detail=f"任务不存在:{task_id}")
data = payload.model_dump(exclude_unset=True)
if "group" in data:
t.group_id = _resolve_group_id(db, data.pop("group"))
for k, v in data.items():
setattr(t, k, v)
try:
db.commit()
except IntegrityError:
db.rollback()
raise HTTPException(status_code=409, detail=f"任务名已存在:{payload.name}")
db.refresh(t)
return _row_to_out(t)
@router.delete("/{task_id}", status_code=204, summary="删除任务")
def delete_task(task_id: int, db: Session = Depends(get_session)):
t = db.query(Task).filter(Task.id == task_id).first()
if not t:
raise HTTPException(status_code=404, detail=f"任务不存在:{task_id}")
db.delete(t)
db.commit()
@router.post("/{task_id}/copy", response_model=TaskOut, summary="复制任务(名字加 _副本)")
def copy_task(task_id: int, db: Session = Depends(get_session)) -> TaskOut:
t = db.query(Task).filter(Task.id == task_id).first()
if not t:
raise HTTPException(status_code=404, detail=f"任务不存在:{task_id}")
# 找不冲突的名字:xxx_副本 / xxx_副本2 / xxx_副本3 ...
base = f"{t.name}_副本"
new_name = base
i = 2
while db.query(Task).filter(Task.name == new_name).first():
new_name = f"{base}{i}"
i += 1
clone = Task(
name=new_name,
group_id=t.group_id,
data_source_path=t.data_source_path,
source_table=t.source_table,
status="停用", # 副本默认停用,避免被定时任务错拉
description=t.description,
db_type=t.db_type,
conn_json=t.conn_json,
)
db.add(clone)
db.commit()
db.refresh(clone)
return _row_to_out(clone)
\ No newline at end of file
/**
* 任务 API 客户端
*
* - GET /api/tasks 列表(group 分组 + keyword 关键字)
* - POST /api/tasks 新建
* - GET /api/tasks/{id} 详情
* - PUT /api/tasks/{id} 更新
* - DELETE /api/tasks/{id} 删除
* - POST /api/tasks/{id}/copy 复制(名字加 _副本)
*/
import { http } from './client'
export function listTasks(params = {}) {
const qs = new URLSearchParams()
if (params.group) qs.set('group', params.group)
if (params.keyword) qs.set('keyword', params.keyword)
const q = qs.toString()
return http.get(`/tasks${q ? `?${q}` : ''}`)
}
export function getTask(id) {
return http.get(`/tasks/${id}`)
}
export function createTask(payload) {
return http.post('/tasks', payload)
}
export function updateTask(id, payload) {
return http.put(`/tasks/${id}`, payload)
}
export function deleteTask(id) {
return http.delete(`/tasks/${id}`)
}
export function copyTask(id) {
return http.post(`/tasks/${id}/copy`)
}
\ No newline at end of file
......@@ -286,19 +286,29 @@ function onSave() {
activeTab.value = 'basic'
return
}
// 后端 P4 才接 fields/rules,P3 只送头部 + conn_json
// (弹窗里改的字段/规则暂存 form,下次保存会带上;当前阶段不写库)
const connJson = JSON.stringify({
connName: form.connName,
host: form.host,
port: form.port,
db: form.db,
user: form.user,
password: form.password,
jdbcParams: form.jdbcParams,
schema: form.schema,
dbType: form.dbType,
table: form.table,
})
const payload = {
name: form.name,
group: form.group,
db: form.db,
table: form.table.split('(')[0],
fields: fieldList.value.length,
rules: fieldList.value.reduce((s, f) => s + (f.rules?.length || 0), 0),
db_type: form.dbType,
data_source_path: null, // P5 接入 data-sources 后再选本地 sqlite
source_table: form.table ? form.table.split('(')[0] : null,
status: '启用',
checkedAt: '2026-08-19 09:00',
tagColor: {
'身份证': 'blue', '手机号': 'orange', '邮箱': 'purple',
'地址信息': 'green', '银行卡号': 'green', '日期格式': 'green',
}[form.group] || 'blue',
description: form.desc,
conn_json: connJson,
}
emit('saved', payload)
}
......
......@@ -8,7 +8,8 @@
<option v-for="g in groups" :key="g" :value="g">{{ g }}</option>
</select>
<span class="query-label">关键字:</span>
<input type="text" class="text-input" placeholder="任务名称/数据表" style="min-width: 200px;" v-model="keyword" />
<input type="text" class="text-input" placeholder="任务名称/数据表" style="min-width: 200px;"
v-model="keyword" @keyup.enter="onQuery" />
<button class="btn btn-primary" @click="onQuery">查询</button>
<button class="btn" @click="onReset">重置</button>
</div>
......@@ -20,7 +21,11 @@
<!-- 任务列表 -->
<div class="q-card">
<div class="q-card__header">
<div class="q-card__title">任务列表<span class="count gray">共 {{ filteredList.length }} 个</span></div>
<div class="q-card__title">任务列表<span class="count gray">共 {{ list.length }} 个</span></div>
<div class="q-card__extra">
<span v-if="loading" style="color:#909399;font-size:13px;">加载中...</span>
<span v-else-if="errorMsg" style="color:#f56c6c;font-size:13px;">{{ errorMsg }}</span>
</div>
</div>
<div class="table-scroll">
<table class="config-list">
......@@ -31,20 +36,23 @@
</tr>
</thead>
<tbody>
<tr v-for="(row, i) in filteredList" :key="i">
<tr v-if="!loading && list.length === 0">
<td colspan="9" style="text-align:center;color:#909399;padding:24px;">暂无任务,点击右上「+ 新增任务」</td>
</tr>
<tr v-for="row in list" :key="row.id">
<td>{{ row.name }}</td>
<td><span class="tag" :class="`tag-${row.tagColor}`">{{ row.group }}</span></td>
<td>{{ row.db }}</td>
<td>{{ row.table }}</td>
<td><span class="tag" :class="`tag-${tagColor(row.group)}`">{{ row.group }}</span></td>
<td>{{ row.db || row.db_type }}</td>
<td>{{ row.source_table || '—' }}</td>
<td>{{ row.fields }}</td>
<td>{{ row.rules }}</td>
<td>
<span class="tag" :class="row.status === '启用' ? 'tag-green' : 'tag-gray'">{{ row.status }}</span>
</td>
<td>{{ row.checkedAt }}</td>
<td>{{ row.checked_at || '—' }}</td>
<td>
<button class="btn-text" @click="openEdit(row)">编辑</button>
<button class="btn-text">复制</button>
<button class="btn-text" @click="onCopy(row)">复制</button>
<button class="btn-text danger" @click="onDelete(row)">删除</button>
</td>
</tr>
......@@ -52,7 +60,7 @@
</table>
</div>
<div class="pagination">
<div class="pagination-info">共 {{ filteredList.length }} 个任务</div>
<div class="pagination-info">共 {{ list.length }} 个任务</div>
<div class="pagination-controls">
<button class="page-btn" disabled>‹</button>
<button class="page-btn active">1</button>
......@@ -66,30 +74,56 @@
</template>
<script setup>
import { computed, inject, ref } from 'vue'
import { computed, inject, onMounted, ref } from 'vue'
import { ElMessage, ElMessageBox } from 'element-plus'
import { taskList } from '@/data/mockData'
import {
listTasks, createTask, updateTask, deleteTask, copyTask,
} from '@/api/tasks'
import TaskModal from '@/components/TaskModal.vue'
const list = ref([...taskList])
const groupFilter = ref('')
const keyword = ref('')
// 分组下拉:从 MainLayout 注入的 taskGroups 码表取(与左侧菜单同源)
const taskGroups = inject('taskGroups', ref([]))
const groups = computed(() => taskGroups.value.map((g) => g.name))
const filteredList = computed(() => {
return list.value.filter((r) => {
if (groupFilter.value && r.group !== groupFilter.value) return false
if (keyword.value && !(r.name.includes(keyword.value) || r.table.includes(keyword.value))) return false
return true
})
})
const list = ref([])
const loading = ref(false)
const errorMsg = ref('')
const groupFilter = ref('')
const keyword = ref('')
// 分组名 → tag 颜色(UI 装饰,与 TaskModal 内置的保持一致)
const GROUP_COLOR = {
'身份证': 'blue', '手机号': 'orange', '邮箱': 'purple',
'地址信息': 'green', '银行卡号': 'green', '日期格式': 'green',
}
function tagColor(g) { return GROUP_COLOR[g] || 'blue' }
// ── 列表 ────────────────────────────────────────────────
async function load() {
loading.value = true
errorMsg.value = ''
try {
list.value = await listTasks({
group: groupFilter.value || undefined,
keyword: keyword.value || undefined,
})
} catch (e) {
errorMsg.value = `加载失败:${e.message}`
} finally {
loading.value = false
}
}
function onQuery() { load() }
function onReset() {
groupFilter.value = ''
keyword.value = ''
load()
}
function onQuery() { /* mock: 过滤已实时生效 */ }
function onReset() { groupFilter.value = ''; keyword.value = '' }
onMounted(load)
// ── 弹窗 ────────────────────────────────────────────────
const modalOpen = ref(false)
const modalMode = ref('add')
const editingTask = ref(null)
......@@ -104,22 +138,47 @@ function openEdit(row) {
editingTask.value = row
modalOpen.value = true
}
function onSaved(payload) {
if (modalMode.value === 'add') {
list.value.unshift(payload)
ElMessage.success('已新增任务')
} else {
const i = list.value.findIndex((r) => r.name === editingTask.value?.name)
if (i >= 0) list.value[i] = { ...list.value[i], ...payload }
ElMessage.success('已保存修改')
async function onSaved(payload) {
try {
if (modalMode.value === 'add') {
const created = await createTask(payload)
list.value.unshift(created)
ElMessage.success('已新增任务')
} else {
const id = editingTask.value?.id
if (!id) throw new Error('缺少任务 id')
const updated = await updateTask(id, payload)
const i = list.value.findIndex((r) => r.id === id)
if (i >= 0) list.value[i] = updated
ElMessage.success('已保存修改')
}
modalOpen.value = false
} catch (e) {
ElMessage.error(`保存失败:${e.message}`)
}
}
async function onCopy(row) {
try {
const cloned = await copyTask(row.id)
list.value.unshift(cloned)
ElMessage.success(`已复制为「${cloned.name}」`)
} catch (e) {
ElMessage.error(`复制失败:${e.message}`)
}
modalOpen.value = false
}
function onDelete(row) {
ElMessageBox.confirm(`确定删除任务「${row.name}」?`, '提示', { type: 'warning' })
.then(() => {
list.value = list.value.filter((r) => r.name !== row.name)
ElMessage.success('已删除')
}).catch(() => {})
async function onDelete(row) {
try {
await ElMessageBox.confirm(`确定删除任务「${row.name}」?`, '提示', { type: 'warning' })
} catch { return }
try {
await deleteTask(row.id)
list.value = list.value.filter((r) => r.id !== row.id)
ElMessage.success('已删除')
} catch (e) {
ElMessage.error(`删除失败:${e.message}`)
}
}
</script>
\ No newline at end of file
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