loganandClaude Opus 4.7
439155c92e
FastAPI backend with auth, user, instance, ticket, query, dashboard APIs. React + TypeScript frontend with mock data UI (10 pages). SQLite + async SQLAlchemy, JWT auth, role-based access control. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2056 lines
67 KiB
Markdown
2056 lines
67 KiB
Markdown
# DataGuard 数据库变更审核平台 — 全面系统性优化指南
|
||
|
||
> 版本: v3.0 | 日期: 2026-05-20
|
||
> 范围: FastAPI 后端 + Vue3 前端，支持 MySQL / PostgreSQL / MongoDB / Redis
|
||
|
||
---
|
||
|
||
## 1. 总体架构优化建议
|
||
|
||
### 1.1 FastAPI 后端最佳项目结构
|
||
|
||
```
|
||
backend/
|
||
├── app/
|
||
│ ├── __init__.py
|
||
│ ├── main.py # FastAPI 应用入口，注册 router/middleware
|
||
│ ├── core/
|
||
│ │ ├── config.py # Pydantic Settings 多环境配置
|
||
│ │ ├── exceptions.py # 全局异常定义
|
||
│ │ ├── security.py # JWT / OAuth2 / 密码哈希
|
||
│ │ ├── permissions.py # RBAC 权限校验装饰器
|
||
│ │ └── database.py # 多数据库连接池管理
|
||
│ ├── api/
|
||
│ │ ├── __init__.py
|
||
│ │ └── v1/
|
||
│ │ ├── __init__.py
|
||
│ │ ├── router.py # APIRouter 聚合
|
||
│ │ ├── endpoints/
|
||
│ │ │ ├── auth.py
|
||
│ │ │ ├── user.py
|
||
│ │ │ ├── role.py
|
||
│ │ │ ├── instance.py
|
||
│ │ │ ├── ticket.py
|
||
│ │ │ ├── query.py
|
||
│ │ │ ├── audit_rule.py
|
||
│ │ │ ├── workflow.py
|
||
│ │ │ ├── config.py
|
||
│ │ │ ├── task.py
|
||
│ │ │ └── dashboard.py
|
||
│ ├── models/
|
||
│ │ ├── __init__.py
|
||
│ │ ├── user.py # SQLAlchemy Model
|
||
│ │ ├── instance.py
|
||
│ │ ├── ticket.py
|
||
│ │ └── audit_log.py
|
||
│ ├── schemas/
|
||
│ │ ├── __init__.py
|
||
│ │ ├── base.py # 通用 Pydantic Schema
|
||
│ │ ├── user.py # Request / Response DTO
|
||
│ │ ├── ticket.py
|
||
│ │ └── instance.py
|
||
│ ├── services/
|
||
│ │ ├── __init__.py
|
||
│ │ ├── ticket_service.py # 业务逻辑层
|
||
│ │ ├── query_service.py
|
||
│ │ ├── instance_service.py
|
||
│ │ └── audit_service.py
|
||
│ ├── crud/
|
||
│ │ ├── __init__.py
|
||
│ │ ├── base.py # 通用 CRUD 基类
|
||
│ │ ├── user.py
|
||
│ │ └── ticket.py
|
||
│ ├── dependencies/
|
||
│ │ ├── __init__.py
|
||
│ │ ├── auth_dep.py # get_current_user / require_role
|
||
│ │ └── db_dep.py # get_db / get_redis
|
||
│ ├── tasks/
|
||
│ │ ├── __init__.py
|
||
│ │ └── celery_tasks.py # 异步任务定义
|
||
│ ├── utils/
|
||
│ │ ├── sql_parser.py # SQL AST 解析
|
||
│ │ ├── sql_rollback.py # Binlog 回滚生成
|
||
│ │ ├── data_mask.py # 数据脱敏
|
||
│ │ ├── notify.py # 多渠道通知
|
||
│ │ └── audit_logger.py # 审计日志记录
|
||
│ └── plugins/
|
||
│ ├── mysql_driver.py # MySQL 连接封装
|
||
│ ├── pg_driver.py # PostgreSQL 连接封装
|
||
│ ├── mongo_driver.py # MongoDB 连接封装
|
||
│ └── redis_driver.py # Redis 连接封装
|
||
├── migrations/ # Alembic 数据库迁移
|
||
├── tests/
|
||
├── Dockerfile
|
||
├── docker-compose.yml
|
||
├── pyproject.toml
|
||
├── alembic.ini
|
||
└── .env / .env.test / .env.prod
|
||
```
|
||
|
||
**核心原则**:
|
||
- APIRouter 按模块拆分，v1/router.py 统一聚合
|
||
- 依赖注入: `Depends(get_db)`, `Depends(require_role("admin"))`
|
||
- Clean Architecture: API -> Service -> CRUD -> Model，禁止跨层调用
|
||
- 异步优先: 所有 IO 操作使用 `async/await`
|
||
|
||
### 1.2 Vue3 前端架构优化
|
||
|
||
```
|
||
frontend/
|
||
├── src/
|
||
│ ├── main.ts
|
||
│ ├── App.vue
|
||
│ ├── router/
|
||
│ │ ├── index.ts # 路由定义 + 路由守卫
|
||
│ │ └── guards.ts # 权限守卫 / 登录守卫
|
||
│ ├── stores/
|
||
│ │ ├── auth.ts # Pinia: 用户/权限/菜单
|
||
│ │ ├── ticket.ts # Pinia: 工单状态
|
||
│ │ ├── instance.ts # Pinia: 数据源状态
|
||
│ │ └── config.ts # Pinia: 全局配置
|
||
│ ├── api/
|
||
│ │ ├── request.ts # axios 封装 + 拦截器
|
||
│ │ ├── auth.ts # API 接口定义
|
||
│ │ ├── ticket.ts
|
||
│ │ ├── instance.ts
|
||
│ │ └── ...
|
||
│ ├── composables/
|
||
│ │ ├── useAuth.ts # 权限相关组合式函数
|
||
│ │ ├── usePermission.ts
|
||
│ │ ├── useTable.ts # 表格分页/筛选/排序通用逻辑
|
||
│ │ └── useNotification.ts
|
||
│ ├── views/
|
||
│ │ ├── dashboard/
|
||
│ │ ├── tickets/
|
||
│ │ ├── queries/
|
||
│ │ ├── instances/
|
||
│ │ ├── workflow/
|
||
│ │ ├── config/
|
||
│ │ └── system/
|
||
│ ├── components/
|
||
│ │ ├── common/ # 通用组件
|
||
│ │ │ ├── PageHeader.vue
|
||
│ │ │ ├── DataTable.vue # 通用表格（分页/排序/筛选）
|
||
│ │ │ ├── FilterBar.vue
|
||
│ │ │ ├── EmptyState.vue
|
||
│ │ │ └── StatCards.vue
|
||
│ │ ├── sql-editor/ # SQL 编辑器相关
|
||
│ │ │ └── MonacoEditor.vue
|
||
│ │ └── workflow/ # 审批流程可视化
|
||
│ │ └── FlowChart.vue
|
||
│ ├── types/
|
||
│ │ └── api.ts # API 类型定义（与 Pydantic 同步）
|
||
│ └── utils/
|
||
│ ├── formatter.ts
|
||
│ └── constants.ts
|
||
```
|
||
|
||
**状态管理策略**:
|
||
- **Pinia**: 全局状态（用户、权限、菜单、配置）
|
||
- **TanStack Query (Vue Query)**: 服务端状态管理，自动缓存/刷新/去重
|
||
|
||
```typescript
|
||
// stores/auth.ts - Pinia 示例
|
||
import { defineStore } from 'pinia';
|
||
|
||
export const useAuthStore = defineStore('auth', {
|
||
state: () => ({
|
||
user: null as User | null,
|
||
token: localStorage.getItem('token') || '',
|
||
permissions: [] as string[],
|
||
}),
|
||
getters: {
|
||
isAdmin: (s) => s.user?.role === 'admin',
|
||
isDBA: (s) => s.user?.role === 'dba',
|
||
},
|
||
actions: {
|
||
async login(credentials: LoginForm) {
|
||
const res = await apiAuth.login(credentials);
|
||
this.user = res.user;
|
||
this.token = res.token;
|
||
this.permissions = res.permissions;
|
||
localStorage.setItem('token', res.token);
|
||
},
|
||
hasPermission(action: string): boolean {
|
||
return this.permissions.includes(action);
|
||
},
|
||
},
|
||
});
|
||
```
|
||
|
||
### 1.3 前后端类型同步方案
|
||
|
||
使用 Pydantic + OpenAPI 自动生成前端类型:
|
||
|
||
```python
|
||
# backend/app/schemas/ticket.py
|
||
from pydantic import BaseModel, Field
|
||
from datetime import datetime
|
||
from enum import Enum
|
||
|
||
class TicketStatus(str, Enum):
|
||
PENDING = "pending"
|
||
REVIEWING = "reviewing"
|
||
APPROVED = "approved"
|
||
REJECTED = "rejected"
|
||
EXECUTING = "executing"
|
||
SUCCESS = "success"
|
||
FAILED = "failed"
|
||
|
||
class TicketCreate(BaseModel):
|
||
title: str = Field(..., min_length=5, max_length=200)
|
||
db_type: str = Field(..., pattern="^(mysql|postgres|mongo|redis)$")
|
||
instance_id: int
|
||
sql_content: str = Field(..., min_length=1)
|
||
rollback_sql: str | None = None
|
||
reason: str = Field(..., min_length=10)
|
||
priority: str = "medium"
|
||
|
||
class TicketResponse(TicketCreate):
|
||
id: int
|
||
status: TicketStatus
|
||
applicant_id: int
|
||
applicant_name: str
|
||
create_time: datetime
|
||
workflow_steps: list[dict] | None = None
|
||
|
||
class Config:
|
||
from_attributes = True
|
||
```
|
||
|
||
前端类型生成命令:
|
||
```bash
|
||
# 使用 openapi-typescript 从 FastAPI OpenAPI 文档自动生成
|
||
npx openapi-typescript http://localhost:8000/openapi.json --output src/types/api.ts
|
||
```
|
||
|
||
### 1.4 多环境隔离与多租户
|
||
|
||
```python
|
||
# backend/app/core/config.py
|
||
from pydantic_settings import BaseSettings
|
||
from functools import lru_cache
|
||
|
||
class Settings(BaseSettings):
|
||
ENV: str = "development" # development / testing / production
|
||
DEBUG: bool = True
|
||
|
||
# JWT
|
||
SECRET_KEY: str
|
||
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
|
||
|
||
# MySQL
|
||
MYSQL_HOST: str = "localhost"
|
||
MYSQL_PORT: int = 3306
|
||
MYSQL_USER: str
|
||
MYSQL_PASSWORD: str
|
||
MYSQL_DB: str
|
||
|
||
# PostgreSQL
|
||
POSTGRES_HOST: str = "localhost"
|
||
POSTGRES_PORT: int = 5432
|
||
POSTGRES_USER: str
|
||
POSTGRES_PASSWORD: str
|
||
POSTGRES_DB: str
|
||
|
||
# MongoDB
|
||
MONGO_URI: str = "mongodb://localhost:27017"
|
||
MONGO_DB: str = "dataguard"
|
||
|
||
# Redis (平台缓存 + Celery Broker)
|
||
REDIS_URL: str = "redis://localhost:6379/0"
|
||
|
||
# Celery
|
||
CELERY_BROKER_URL: str = "redis://localhost:6379/1"
|
||
CELERY_RESULT_BACKEND: str = "redis://localhost:6379/2"
|
||
|
||
class Config:
|
||
env_file = ".env"
|
||
env_file_encoding = "utf-8"
|
||
|
||
@lru_cache()
|
||
def get_settings() -> Settings:
|
||
return Settings()
|
||
```
|
||
|
||
多租户方案:
|
||
- **共享数据库 + 租户ID隔离**: 所有表增加 `tenant_id` 字段
|
||
- **Schema 隔离**: PostgreSQL 支持每租户一个 Schema
|
||
- **数据源隔离**: 每租户独立的数据源配置表，操作自带 `tenant_id` 过滤
|
||
|
||
```python
|
||
# 全局租户上下文
|
||
from contextvars import ContextVar
|
||
|
||
tenant_ctx: ContextVar[int] = ContextVar('tenant_id', default=0)
|
||
|
||
# CRUD 基类自动注入 tenant_id 过滤
|
||
class CRUDBase:
|
||
async def get_multi(self, db: AsyncSession, *, skip=0, limit=100):
|
||
tenant_id = tenant_ctx.get()
|
||
stmt = select(self.model).where(self.model.tenant_id == tenant_id)
|
||
...
|
||
```
|
||
|
||
---
|
||
|
||
## 2. 核心功能模块深度优化
|
||
|
||
### 2.1 数据源管理
|
||
|
||
**当前不足**: 前端展示为主，缺少后端连接池管理、健康检查、自动重连。
|
||
|
||
**优化点**:
|
||
|
||
| 优化项 | 说明 |
|
||
|--------|------|
|
||
| 连接池管理 | 每实例独立连接池，支持 max/min 连接数配置 |
|
||
| 健康检查 | 每 30s 探活，连续 3 次失败标记为断开 |
|
||
| 自动重连 | 指数退避策略重连 |
|
||
| 密码加密 | 实例密码 AES 加密存储 |
|
||
| 连接复用 | 查询审计复用只读连接，工单执行使用独立连接 |
|
||
|
||
**代码结构**:
|
||
|
||
```python
|
||
# backend/app/core/database.py
|
||
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
||
from motor.motor_asyncio import AsyncIOMotorClient
|
||
import aioredis
|
||
|
||
class ConnectionPoolManager:
|
||
_pools: dict[int, Any] = {} # instance_id -> engine/client
|
||
|
||
async def get_mysql_engine(self, instance_id: int):
|
||
if instance_id not in self._pools:
|
||
cfg = await self._get_instance_config(instance_id)
|
||
engine = create_async_engine(
|
||
f"mysql+aiomysql://{cfg.user}:{cfg.password}@{cfg.host}:{cfg.port}/{cfg.db}",
|
||
pool_size=5, max_overflow=10, pool_recycle=3600,
|
||
echo=False,
|
||
)
|
||
self._pools[instance_id] = engine
|
||
return self._pools[instance_id]
|
||
|
||
async def get_mongo_client(self, instance_id: int):
|
||
if instance_id not in self._pools:
|
||
cfg = await self._get_instance_config(instance_id)
|
||
client = AsyncIOMotorClient(
|
||
f"mongodb://{cfg.user}:{cfg.password}@{cfg.host}:{cfg.port}",
|
||
maxPoolSize=10, minPoolSize=1,
|
||
)
|
||
self._pools[instance_id] = client
|
||
return client
|
||
|
||
async def health_check(self, instance_id: int) -> bool:
|
||
"""探活检查，连续失败3次标记 offline"""
|
||
...
|
||
|
||
pool_mgr = ConnectionPoolManager()
|
||
```
|
||
|
||
### 2.2 SQL 工单全生命周期
|
||
|
||
**当前不足**: 审批流程硬编码，缺少自动审核引擎、回滚 SQL 自动生成。
|
||
|
||
**优化点**:
|
||
|
||
```
|
||
提交(Submit)
|
||
-> 自动审核(AutoAudit: 语法检查 + 规则匹配 + 影响分析)
|
||
-> 人工审批(ManualReview: 多级审批流)
|
||
-> 执行(Execute: 预检查 + 执行 + 结果记录)
|
||
-> 回滚(Rollback: Binlog逆向 + 备份恢复)
|
||
```
|
||
|
||
**状态机设计**:
|
||
|
||
```python
|
||
# backend/app/services/ticket_fsm.py
|
||
from enum import Enum, auto
|
||
|
||
class TicketState(str, Enum):
|
||
DRAFT = "draft"
|
||
AUTO_AUDITING = "auto_auditing"
|
||
AUTO_AUDIT_FAILED = "auto_audit_failed"
|
||
PENDING_REVIEW = "pending_review"
|
||
REVIEWING = "reviewing"
|
||
APPROVED = "approved"
|
||
REJECTED = "rejected"
|
||
PRE_EXECUTE = "pre_execute"
|
||
EXECUTING = "executing"
|
||
SUCCESS = "success"
|
||
FAILED = "failed"
|
||
ROLLING_BACK = "rolling_back"
|
||
ROLLED_BACK = "rolled_back"
|
||
|
||
class TicketFSM:
|
||
TRANSITIONS = {
|
||
TicketState.DRAFT: [TicketState.AUTO_AUDITING],
|
||
TicketState.AUTO_AUDITING: [TicketState.PENDING_REVIEW, TicketState.AUTO_AUDIT_FAILED],
|
||
TicketState.AUTO_AUDIT_FAILED: [TicketState.DRAFT], # 可修改重提
|
||
TicketState.PENDING_REVIEW: [TicketState.REVIEWING],
|
||
TicketState.REVIEWING: [TicketState.APPROVED, TicketState.REJECTED],
|
||
TicketState.APPROVED: [TicketState.PRE_EXECUTE],
|
||
TicketState.PRE_EXECUTE: [TicketState.EXECUTING],
|
||
TicketState.EXECUTING: [TicketState.SUCCESS, TicketState.FAILED],
|
||
TicketState.FAILED: [TicketState.ROLLING_BACK],
|
||
TicketState.ROLLING_BACK: [TicketState.ROLLED_BACK],
|
||
}
|
||
|
||
def can_transition(self, current: TicketState, target: TicketState) -> bool:
|
||
return target in self.TRANSITIONS.get(current, [])
|
||
|
||
async def transition(self, ticket_id: int, target: TicketState, operator_id: int):
|
||
ticket = await crud.ticket.get(db, ticket_id)
|
||
if not self.can_transition(ticket.status, target):
|
||
raise BusinessException(f"非法状态流转: {ticket.status} -> {target}")
|
||
...
|
||
```
|
||
|
||
**自动审核引擎**:
|
||
|
||
```python
|
||
# backend/app/services/auto_audit.py
|
||
class AutoAuditEngine:
|
||
"""自动审核引擎：语法解析 + 规则匹配 + 影响分析"""
|
||
|
||
async def audit(self, ticket_id: int) -> AuditResult:
|
||
ticket = await crud.ticket.get(db, ticket_id)
|
||
|
||
# 1. SQL 语法解析
|
||
statements = sqlparse.parse(ticket.sql_content)
|
||
for stmt in statements:
|
||
if stmt.get_type() == 'UNKNOWN':
|
||
return AuditResult(passed=False, score=0, issues=["SQL 语法错误"])
|
||
|
||
# 2. 规则匹配
|
||
rules = await crud.audit_rule.get_active_rules(db, db_type=ticket.db_type)
|
||
issues = []
|
||
for rule in rules:
|
||
result = await self._check_rule(rule, ticket.sql_content)
|
||
if not result.passed:
|
||
issues.append(result)
|
||
|
||
# 3. 影响行数估算 (MySQL/PostgreSQL)
|
||
if ticket.db_type in ('mysql', 'postgres'):
|
||
explain = await self._explain_sql(ticket)
|
||
estimated_rows = explain.get('rows', 0)
|
||
if estimated_rows > 10000:
|
||
issues.append(AuditIssue(
|
||
level='warning',
|
||
message=f'预估影响 {estimated_rows} 行，建议分批执行'
|
||
))
|
||
|
||
# 4. 生成回滚 SQL (MySQL Binlog)
|
||
if ticket.db_type == 'mysql':
|
||
rollback_sql = await self._generate_rollback(ticket)
|
||
await crud.ticket.update(db, ticket_id, rollback_sql=rollback_sql)
|
||
|
||
score = max(0, 100 - len([i for i in issues if i.level == 'error']) * 20
|
||
- len([i for i in issues if i.level == 'warning']) * 5)
|
||
|
||
return AuditResult(
|
||
passed=score >= 60 and not any(i.level == 'error' for i in issues),
|
||
score=score,
|
||
issues=issues,
|
||
)
|
||
```
|
||
|
||
**回滚 SQL 生成 (MySQL)**:
|
||
|
||
```python
|
||
# backend/app/utils/sql_rollback.py
|
||
class RollbackGenerator:
|
||
"""基于 Binlog ROW 格式生成逆向 SQL"""
|
||
|
||
async def generate_for_update(self, instance_id: int, table: str,
|
||
where_clause: str) -> str:
|
||
"""
|
||
1. 连接 MySQL 读取 Binlog（或直接从表读取变更前数据）
|
||
2. 使用 PRIMARY KEY 定位行
|
||
3. 生成 UPDATE ... SET col=old_val WHERE pk=val
|
||
"""
|
||
engine = await pool_mgr.get_mysql_engine(instance_id)
|
||
async with engine.connect() as conn:
|
||
# 获取表结构 + 主键
|
||
pk = await self._get_primary_key(conn, table)
|
||
|
||
# 读取变更前数据（通过 Flashback 或备份表）
|
||
backup_table = f"_dg_backup_{table}_{int(time.time())}"
|
||
await conn.execute(text(f"CREATE TABLE {backup_table} AS SELECT * FROM {table} WHERE {where_clause}"))
|
||
|
||
# 生成回滚 SQL
|
||
rollback = f"""-- 回滚 SQL（基于 Binlog ROW 解析）
|
||
-- 备份表: {backup_table}
|
||
-- 恢复方式: 从备份表还原
|
||
INSERT INTO {table} SELECT * FROM {backup_table}
|
||
ON DUPLICATE KEY UPDATE
|
||
c1=VALUES(c1), c2=VALUES(c2);
|
||
"""
|
||
return rollback
|
||
|
||
async def generate_for_delete(self, instance_id: int, table: str,
|
||
where_clause: str) -> str:
|
||
"""DELETE 回滚 = INSERT 被删数据"""
|
||
...
|
||
```
|
||
|
||
### 2.3 在线 SQL 查询器
|
||
|
||
**当前不足**: textarea 编辑，无语法高亮、无自动补全。
|
||
|
||
**优化方案**:
|
||
|
||
```typescript
|
||
// frontend/src/components/sql-editor/MonacoEditor.vue
|
||
<template>
|
||
<div ref="editorRef" class="sql-editor" />
|
||
</template>
|
||
|
||
<script setup lang="ts">
|
||
import { ref, onMounted, watch } from 'vue';
|
||
import * as monaco from 'monaco-editor';
|
||
import { setupMySQLLanguage } from './mysql-language';
|
||
import { setupPGLanguage } from './pg-language';
|
||
|
||
const props = defineProps<{
|
||
modelValue: string;
|
||
dbType: 'mysql' | 'postgres' | 'mongo' | 'redis';
|
||
}>();
|
||
|
||
const editorRef = ref<HTMLElement>();
|
||
let editor: monaco.editor.IStandaloneCodeEditor;
|
||
|
||
onMounted(() => {
|
||
// 注册数据库语法高亮
|
||
if (props.dbType === 'mysql') setupMySQLLanguage();
|
||
if (props.dbType === 'postgres') setupPGLanguage();
|
||
|
||
editor = monaco.editor.create(editorRef.value!, {
|
||
value: props.modelValue,
|
||
language: props.dbType === 'mysql' ? 'mysql' : props.dbType === 'postgres' ? 'pgsql' : 'sql',
|
||
theme: 'vs-dark',
|
||
minimap: { enabled: false },
|
||
automaticLayout: true,
|
||
fontSize: 13,
|
||
fontFamily: 'JetBrains Mono, monospace',
|
||
suggestOnTriggerCharacters: true,
|
||
quickSuggestions: true,
|
||
wordBasedSuggestions: true,
|
||
});
|
||
|
||
// 自动补全：表名/字段名
|
||
monaco.languages.registerCompletionItemProvider('mysql', {
|
||
triggerCharacters: ['.', ' '],
|
||
provideCompletionItems: async (model, position) => {
|
||
// 从后端获取当前库的表名和字段名
|
||
const suggestions = await fetchSchemaSuggestions(props.dbType);
|
||
return { suggestions };
|
||
},
|
||
});
|
||
});
|
||
</script>
|
||
```
|
||
|
||
**后端查询执行**:
|
||
|
||
```python
|
||
# backend/app/services/query_service.py
|
||
class QueryService:
|
||
async def execute_query(self, instance_id: int, sql: str, user_id: int,
|
||
max_rows: int = 10000) -> QueryResult:
|
||
instance = await crud.instance.get(db, instance_id)
|
||
|
||
# 1. 权限检查
|
||
if not await self._can_query(user_id, instance_id):
|
||
raise PermissionDenied("无权访问该数据源")
|
||
|
||
# 2. SQL 拦截检查
|
||
forbidden_keywords = ['DROP', 'TRUNCATE', 'DELETE', 'UPDATE', 'INSERT', 'CREATE', 'ALTER']
|
||
upper_sql = sql.strip().upper()
|
||
if any(kw in upper_sql for kw in forbidden_keywords):
|
||
raise BusinessException("查询模式禁止执行写操作")
|
||
|
||
# 3. 添加 LIMIT
|
||
if not re.search(r'LIMIT\s+\d+', upper_sql):
|
||
sql = f"{sql.rstrip(';')} LIMIT {max_rows};"
|
||
|
||
# 4. 执行（带超时）
|
||
start = time.time()
|
||
engine = await pool_mgr.get_mysql_engine(instance_id)
|
||
async with engine.connect() as conn:
|
||
result = await asyncio.wait_for(
|
||
conn.execute(text(sql)), timeout=30
|
||
)
|
||
rows = result.mappings().all()
|
||
columns = list(result.keys())
|
||
elapsed = int((time.time() - start) * 1000)
|
||
|
||
# 5. 敏感数据脱敏
|
||
masked_rows = await self._mask_sensitive_data(instance_id, columns, rows)
|
||
|
||
# 6. 记录审计日志
|
||
await audit_logger.log_query(user_id, instance_id, sql, elapsed, len(rows))
|
||
|
||
return QueryResult(columns=columns, rows=masked_rows, elapsed_ms=elapsed)
|
||
```
|
||
|
||
### 2.4 数据字典
|
||
|
||
【补充需求】需确认是否需要自动采集数据字典（从 information_schema 读取）还是手动维护。
|
||
|
||
```python
|
||
# backend/app/services/data_dict.py
|
||
class DataDictionaryService:
|
||
async def sync_mysql_schema(self, instance_id: int):
|
||
"""从 MySQL information_schema 自动同步表结构"""
|
||
engine = await pool_mgr.get_mysql_engine(instance_id)
|
||
async with engine.connect() as conn:
|
||
# 获取所有表
|
||
tables = await conn.execute(text("""
|
||
SELECT TABLE_NAME, TABLE_COMMENT
|
||
FROM information_schema.TABLES
|
||
WHERE TABLE_SCHEMA = DATABASE()
|
||
"""))
|
||
|
||
for table in tables.mappings():
|
||
# 获取字段
|
||
columns = await conn.execute(text(f"""
|
||
SELECT COLUMN_NAME, DATA_TYPE, COLUMN_COMMENT,
|
||
IS_NULLABLE, COLUMN_DEFAULT, COLUMN_KEY
|
||
FROM information_schema.COLUMNS
|
||
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '{table.TABLE_NAME}'
|
||
"""))
|
||
|
||
# 保存到数据字典表
|
||
await crud.data_dict.save_table(instance_id, table, columns)
|
||
```
|
||
|
||
### 2.5 Redis 专用管理功能
|
||
|
||
**当前不足**: Redis 与关系数据库混用同一套界面，缺少 Redis 特有的管理功能。
|
||
|
||
**优化点**:
|
||
|
||
| 功能 | 说明 |
|
||
|------|------|
|
||
| Key 浏览 | 树形结构展示，支持按 pattern 过滤 |
|
||
| Key 操作 | GET / SET / DEL / EXPIRE / TTL |
|
||
| 数据结构支持 | String / Hash / List / Set / ZSet |
|
||
| 实时监控 | 内存使用、命中率、QPS、慢查询 |
|
||
| 批量操作 | 批量删除、批量设置 TTL |
|
||
| 序列化识别 | 自动识别 JSON / Protobuf / 压缩数据 |
|
||
|
||
```python
|
||
# backend/app/plugins/redis_driver.py
|
||
class RedisManager:
|
||
async def get_keys(self, instance_id: int, pattern: str = "*",
|
||
cursor: int = 0, count: int = 100) -> dict:
|
||
client = await pool_mgr.get_redis_client(instance_id)
|
||
cur, keys = await client.scan(cursor=cursor, match=pattern, count=count)
|
||
|
||
results = []
|
||
for key in keys:
|
||
key_type = await client.type(key)
|
||
ttl = await client.ttl(key)
|
||
memory = await client.memory_usage(key) or 0
|
||
results.append({
|
||
'key': key.decode(),
|
||
'type': key_type.decode(),
|
||
'ttl': ttl,
|
||
'memory': memory,
|
||
})
|
||
return {'cursor': cur, 'keys': results}
|
||
|
||
async def get_value(self, instance_id: int, key: str) -> dict:
|
||
"""根据类型返回格式化后的值"""
|
||
client = await pool_mgr.get_redis_client(instance_id)
|
||
key_type = (await client.type(key)).decode()
|
||
|
||
handlers = {
|
||
'string': lambda: client.get(key),
|
||
'hash': lambda: client.hgetall(key),
|
||
'list': lambda: client.lrange(key, 0, -1),
|
||
'set': lambda: client.smembers(key),
|
||
'zset': lambda: client.zrange(key, 0, -1, withscores=True),
|
||
}
|
||
|
||
raw = await handlers[key_type]()
|
||
return {'type': key_type, 'value': self._deserialize(raw)}
|
||
|
||
def _deserialize(self, raw) -> Any:
|
||
"""自动识别并反序列化 JSON / 压缩数据"""
|
||
if isinstance(raw, bytes):
|
||
try:
|
||
return json.loads(raw.decode('utf-8'))
|
||
except (json.JSONDecodeError, UnicodeDecodeError):
|
||
return raw.decode('utf-8', errors='replace')
|
||
if isinstance(raw, dict):
|
||
return {k.decode() if isinstance(k, bytes) else k:
|
||
self._deserialize(v) for k, v in raw.items()}
|
||
if isinstance(raw, (list, set, tuple)):
|
||
return [self._deserialize(v) for v in raw]
|
||
return raw
|
||
```
|
||
|
||
### 2.6 仪表盘与监控
|
||
|
||
**优化点**:
|
||
- 工单趋势（日/周/月）
|
||
- 状态分布饼图
|
||
- 慢查询 TOP10
|
||
- 数据源健康状态
|
||
- 待办审批提醒
|
||
|
||
```python
|
||
# backend/app/services/dashboard.py
|
||
class DashboardService:
|
||
async def get_stats(self, tenant_id: int) -> DashboardStats:
|
||
# 使用缓存加速
|
||
cache_key = f"dashboard:stats:{tenant_id}"
|
||
cached = await redis.get(cache_key)
|
||
if cached:
|
||
return DashboardStats.parse_raw(cached)
|
||
|
||
stats = DashboardStats(
|
||
ticket_total=await crud.ticket.count(db, tenant_id=tenant_id),
|
||
ticket_pending=await crud.ticket.count(db, status='pending', tenant_id=tenant_id),
|
||
ticket_month=await crud.ticket.count(db,
|
||
create_time__gte=datetime.now() - timedelta(days=30), tenant_id=tenant_id),
|
||
ticket_executing=await crud.ticket.count(db, status='executing', tenant_id=tenant_id),
|
||
rules_total=await crud.audit_rule.count(db, tenant_id=tenant_id),
|
||
rules_active=await crud.audit_rule.count(db, enabled=True, tenant_id=tenant_id),
|
||
queries_today=await crud.query_audit.count(db,
|
||
timestamp__gte=datetime.now() - timedelta(days=1), tenant_id=tenant_id),
|
||
)
|
||
|
||
await redis.setex(cache_key, 60, stats.json())
|
||
return stats
|
||
```
|
||
|
||
### 2.7 审核规则引擎
|
||
|
||
**当前不足**: 规则硬编码，不支持自定义和版本控制。
|
||
|
||
**优化方案**: 规则引擎支持自定义 + 版本控制。
|
||
|
||
```python
|
||
# backend/app/models/audit_rule.py
|
||
class AuditRule(Base):
|
||
__tablename__ = "audit_rules"
|
||
|
||
id = Column(Integer, primary_key=True)
|
||
name = Column(String(100), nullable=False)
|
||
db_type = Column(String(20), nullable=False) # mysql / postgres / mongo / redis / all
|
||
rule_type = Column(String(30), nullable=False) # regex / ast / risk / custom
|
||
pattern = Column(Text) # 正则/AST模式
|
||
action = Column(String(20), default="warn") # block / warn / require_approval
|
||
description = Column(Text)
|
||
enabled = Column(Boolean, default=True)
|
||
version = Column(Integer, default=1) # 版本号
|
||
priority = Column(Integer, default=100) # 优先级，越小越先匹配
|
||
created_by = Column(Integer, ForeignKey("users.id"))
|
||
created_at = Column(DateTime, default=datetime.now)
|
||
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
|
||
|
||
# 版本历史
|
||
history = relationship("AuditRuleHistory", back_populates="rule")
|
||
|
||
class AuditRuleHistory(Base):
|
||
__tablename__ = "audit_rule_history"
|
||
|
||
id = Column(Integer, primary_key=True)
|
||
rule_id = Column(Integer, ForeignKey("audit_rules.id"))
|
||
version = Column(Integer)
|
||
pattern = Column(Text)
|
||
action = Column(String(20))
|
||
changed_by = Column(Integer)
|
||
changed_at = Column(DateTime, default=datetime.now)
|
||
change_reason = Column(Text)
|
||
```
|
||
|
||
**规则引擎执行**:
|
||
|
||
```python
|
||
# backend/app/services/rule_engine.py
|
||
class RuleEngine:
|
||
async def check(self, sql: str, db_type: str) -> RuleCheckResult:
|
||
rules = await crud.audit_rule.get_active(db, db_type=db_type,
|
||
order_by=AuditRule.priority)
|
||
violations = []
|
||
|
||
for rule in rules:
|
||
matched = False
|
||
|
||
if rule.rule_type == 'regex':
|
||
matched = re.search(rule.pattern, sql, re.IGNORECASE)
|
||
|
||
elif rule.rule_type == 'ast':
|
||
matched = self._check_ast(sql, rule.pattern)
|
||
|
||
elif rule.rule_type == 'risk':
|
||
matched = await self._check_risk(sql, db_type, rule.pattern)
|
||
|
||
elif rule.rule_type == 'custom':
|
||
# 调用用户自定义 Python 函数
|
||
matched = await self._exec_custom_rule(rule.pattern, sql, db_type)
|
||
|
||
if matched:
|
||
violations.append(RuleViolation(
|
||
rule_id=rule.id,
|
||
rule_name=rule.name,
|
||
action=rule.action,
|
||
message=rule.description,
|
||
))
|
||
|
||
if rule.action == 'block':
|
||
break # 拦截规则直接终止
|
||
|
||
return RuleCheckResult(
|
||
passed=not any(v.action == 'block' for v in violations),
|
||
violations=violations,
|
||
)
|
||
```
|
||
|
||
**内置规则列表**:
|
||
|
||
| 规则ID | 名称 | 类型 | 动作 | 适用数据库 |
|
||
|--------|------|------|------|----------|
|
||
| R001 | 禁止无 WHERE 的 UPDATE/DELETE | ast | block | mysql, postgres |
|
||
| R002 | 禁止 SELECT * | regex | warn | mysql, postgres |
|
||
| R003 | 禁止 DROP TABLE/DATABASE | regex | block | mysql, postgres |
|
||
| R004 | 禁止 TRUNCATE TABLE | regex | block | mysql, postgres |
|
||
| R005 | 大表变更检查 (>100万行) | risk | warn | mysql, postgres |
|
||
| R006 | 禁止执行时间超过 10s | risk | block | mysql, postgres |
|
||
| R007 | MongoDB 禁止 db.dropDatabase() | regex | block | mongo |
|
||
| R008 | Redis 禁止 FLUSHALL/FLUSHDB | regex | block | redis |
|
||
| R009 | 禁止无索引查询 (type=ALL) | risk | warn | mysql, postgres |
|
||
| R010 | UPDATE/DELETE 必须提供回滚 SQL | ast | require_approval | mysql, postgres |
|
||
|
||
---
|
||
|
||
## 3. 安全与合规强化
|
||
|
||
### 3.1 SQL 注入防护
|
||
|
||
```python
|
||
# 后端：严格使用参数化查询，禁止字符串拼接 SQL
|
||
# ✅ 正确
|
||
result = await conn.execute(
|
||
text("SELECT * FROM users WHERE id = :user_id"),
|
||
{"user_id": user_id}
|
||
)
|
||
|
||
# ❌ 错误 — 绝对禁止
|
||
result = await conn.execute(f"SELECT * FROM users WHERE id = {user_id}")
|
||
|
||
# 前端：所有用户输入做转义
|
||
import html
|
||
escaped_input = html.escape(user_input)
|
||
```
|
||
|
||
### 3.2 不同数据库的权限最小化
|
||
|
||
| 数据库 | 查询账号 | 执行账号 |
|
||
|--------|---------|---------|
|
||
| MySQL | `SELECT`, `SHOW` | `SELECT, INSERT, UPDATE, DELETE` + 特定表 |
|
||
| PostgreSQL | `CONNECT`, `USAGE`, `SELECT` | 按需 schema 级别授权 |
|
||
| MongoDB | `find` | `insert`, `update`, `remove` + 特定 collection |
|
||
| Redis | 只读实例 / ACL 限制命令集 | 按需配置可执行命令 |
|
||
|
||
```python
|
||
# 执行前权限校验
|
||
async def check_execute_permission(instance_id: int, sql: str, user_id: int):
|
||
# 1. 解析 SQL 类型
|
||
stmt_type = sqlparse.parse(sql)[0].get_type()
|
||
|
||
# 2. 检查用户是否有该实例的执行权限
|
||
if not await rbac.check_instance_permission(user_id, instance_id, stmt_type):
|
||
raise PermissionDenied(f"无权在实例 {instance_id} 上执行 {stmt_type}")
|
||
|
||
# 3. 行级权限检查
|
||
tables = extract_tables(sql)
|
||
for table in tables:
|
||
if not await rbac.check_table_permission(user_id, instance_id, table):
|
||
raise PermissionDenied(f"无权访问表 {table}")
|
||
```
|
||
|
||
### 3.3 敏感数据脱敏
|
||
|
||
```python
|
||
# backend/app/utils/data_mask.py
|
||
class DataMasker:
|
||
RULES = [
|
||
{'pattern': r'^1[3-9]\d{9}$', 'type': 'phone', 'mask': lambda v: v[:3]+'****'+v[-4:]},
|
||
{'pattern': r'^[\w.-]+@[\w.-]+\.\w+$', 'type': 'email', 'mask': lambda v: v[:3]+'***@'+v.split('@')[1]},
|
||
{'pattern': r'^\d{17}[\dXx]$', 'type': 'id_card', 'mask': lambda v: v[:6]+'********'+v[-4:]},
|
||
{'pattern': r'^\d{16,19}$', 'type': 'bank_card', 'mask': lambda v: '**** **** **** '+v[-4:]},
|
||
]
|
||
|
||
COLUMN_PATTERNS = {
|
||
'phone': ['phone', 'mobile', 'tel'],
|
||
'email': ['email', 'e_mail', 'mail'],
|
||
'id_card': ['id_card', 'idcard', 'identity', 'ssn'],
|
||
'bank_card': ['bank_card', 'bankcard', 'card_no'],
|
||
}
|
||
|
||
async def mask_value(self, column_name: str, value: str) -> str:
|
||
col_lower = column_name.lower()
|
||
for data_type, keywords in self.COLUMN_PATTERNS.items():
|
||
if any(kw in col_lower for kw in keywords):
|
||
rule = next(r for r in self.RULES if r['type'] == data_type)
|
||
return rule['mask'](str(value))
|
||
return value
|
||
|
||
async def mask_result(self, columns: list[str], rows: list[dict]) -> list[dict]:
|
||
to_mask = {i: None for i, col in enumerate(columns)
|
||
if any(kw in col.lower() for kw in sum(self.COLUMN_PATTERNS.values(), []))}
|
||
|
||
for row in rows:
|
||
for col_idx in to_mask:
|
||
col_name = columns[col_idx]
|
||
val = str(row.get(col_name, ''))
|
||
for rule in self.RULES:
|
||
if re.match(rule['pattern'], val):
|
||
row[col_name] = rule['mask'](val)
|
||
break
|
||
return rows
|
||
```
|
||
|
||
### 3.4 操作审计日志（含变更前后对比）
|
||
|
||
```python
|
||
# backend/app/utils/audit_logger.py
|
||
class AuditLogger:
|
||
async def log_change(self, user_id: int, action: str, target_type: str,
|
||
target_id: int, before: dict, after: dict):
|
||
"""记录变更操作，包含前后数据对比"""
|
||
diff = self._compute_diff(before, after)
|
||
|
||
log = AuditLog(
|
||
user_id=user_id,
|
||
action=action, # CREATE / UPDATE / DELETE / EXECUTE
|
||
target_type=target_type, # ticket / instance / user / query
|
||
target_id=target_id,
|
||
before_data=json.dumps(before, ensure_ascii=False, default=str),
|
||
after_data=json.dumps(after, ensure_ascii=False, default=str),
|
||
diff=json.dumps(diff, ensure_ascii=False),
|
||
ip_address=RequestContext.get_ip(),
|
||
user_agent=RequestContext.get_ua(),
|
||
created_at=datetime.now(),
|
||
)
|
||
await db.add(log)
|
||
|
||
def _compute_diff(self, before: dict, after: dict) -> list[dict]:
|
||
"""计算变更差异"""
|
||
diff = []
|
||
all_keys = set(before.keys()) | set(after.keys())
|
||
for key in all_keys:
|
||
old_val = before.get(key)
|
||
new_val = after.get(key)
|
||
if old_val != new_val:
|
||
diff.append({
|
||
'field': key,
|
||
'old': old_val,
|
||
'new': new_val,
|
||
})
|
||
return diff
|
||
|
||
# 使用示例
|
||
@AuditLogger.track("工单审批")
|
||
async def approve_ticket(ticket_id: int, user_id: int, comment: str):
|
||
before = await crud.ticket.get(db, ticket_id)
|
||
after = await crud.ticket.update(db, ticket_id, status='approved', approver_id=user_id)
|
||
await audit_logger.log_change(user_id, 'APPROVE', 'ticket', ticket_id,
|
||
before.__dict__, after.__dict__)
|
||
```
|
||
|
||
### 3.5 RBAC + 行级/字段级数据权限
|
||
|
||
```python
|
||
# backend/app/core/permissions.py
|
||
class RBAC:
|
||
# ========== 行级权限 ==========
|
||
async def check_instance_permission(self, user_id: int, instance_id: int) -> bool:
|
||
"""用户是否有该实例的访问权限"""
|
||
user = await crud.user.get(db, user_id)
|
||
if user.role == 'admin':
|
||
return True
|
||
# 检查用户-实例绑定表
|
||
return await crud.user_instance.exists(db, user_id=user_id, instance_id=instance_id)
|
||
|
||
# ========== 字段级权限 ==========
|
||
async def filter_columns(self, user_id: int, instance_id: int,
|
||
table: str, columns: list[str]) -> list[str]:
|
||
"""过滤用户无权查看的字段"""
|
||
user = await crud.user.get(db, user_id)
|
||
if user.role in ('admin', 'dba'):
|
||
return columns
|
||
|
||
# 查询字段权限
|
||
allowed = await crud.column_permission.get_allowed(
|
||
db, user_id=user_id, instance_id=instance_id, table=table
|
||
)
|
||
return [c for c in columns if c in allowed]
|
||
|
||
# ========== 数据范围权限 ==========
|
||
async def get_data_scope(self, user_id: int) -> DataScope:
|
||
"""获取用户的数据范围权限（全部/本部门/本人）"""
|
||
user = await crud.user.get(db, user_id)
|
||
scope_map = {
|
||
'admin': DataScope.ALL,
|
||
'dba': DataScope.ALL,
|
||
'developer': DataScope.SELF,
|
||
'guest': DataScope.SELF,
|
||
}
|
||
return scope_map.get(user.role, DataScope.NONE)
|
||
|
||
# 装饰器封装
|
||
from functools import wraps
|
||
|
||
def require_role(*roles: str):
|
||
def decorator(func):
|
||
@wraps(func)
|
||
async def wrapper(*args, **kwargs):
|
||
current_user = kwargs.get('current_user')
|
||
if not current_user or current_user.role not in roles:
|
||
raise PermissionDenied(f"需要角色: {', '.join(roles)}")
|
||
return await func(*args, **kwargs)
|
||
return wrapper
|
||
return decorator
|
||
|
||
# API 使用
|
||
@router.post("/tickets/{ticket_id}/approve")
|
||
async def approve_ticket(
|
||
ticket_id: int,
|
||
current_user: User = Depends(get_current_user),
|
||
):
|
||
# 自动校验 role in ('admin', 'dba')
|
||
require_role('admin', 'dba')
|
||
...
|
||
```
|
||
|
||
### 3.6 JWT + OAuth2 最佳实践
|
||
|
||
```python
|
||
# backend/app/core/security.py
|
||
from jose import jwt, JWTError
|
||
from passlib.context import CryptContext
|
||
|
||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||
|
||
class SecurityManager:
|
||
ALGORITHM = "HS256"
|
||
ACCESS_TOKEN_EXPIRE = 30 # minutes
|
||
REFRESH_TOKEN_EXPIRE = 7 # days
|
||
|
||
@staticmethod
|
||
def verify_password(plain: str, hashed: str) -> bool:
|
||
return pwd_context.verify(plain, hashed)
|
||
|
||
@staticmethod
|
||
def hash_password(password: str) -> str:
|
||
return pwd_context.hash(password)
|
||
|
||
@classmethod
|
||
def create_token(cls, user_id: int, token_type: str = "access") -> str:
|
||
expire = datetime.utcnow() + timedelta(
|
||
minutes=cls.ACCESS_TOKEN_EXPIRE if token_type == "access" else cls.REFRESH_TOKEN_EXPIRE
|
||
)
|
||
payload = {
|
||
"sub": str(user_id),
|
||
"type": token_type,
|
||
"exp": expire,
|
||
"iat": datetime.utcnow(),
|
||
"jti": secrets.token_urlsafe(16), # 唯一标识，用于撤销
|
||
}
|
||
return jwt.encode(payload, settings.SECRET_KEY, algorithm=cls.ALGORITHM)
|
||
|
||
@classmethod
|
||
async def verify_token(cls, token: str) -> TokenPayload:
|
||
try:
|
||
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[cls.ALGORITHM])
|
||
# 检查 Token 是否在撤销列表
|
||
if await redis.sismember("revoked_tokens", payload.get("jti")):
|
||
raise JWTError("Token 已撤销")
|
||
return TokenPayload(**payload)
|
||
except JWTError as e:
|
||
raise AuthenticationFailed(f"无效的 Token: {e}")
|
||
|
||
# 依赖注入
|
||
async def get_current_user(
|
||
token: str = Depends(OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login")),
|
||
) -> User:
|
||
payload = await SecurityManager.verify_token(token)
|
||
user = await crud.user.get(db, int(payload.sub))
|
||
if not user or not user.is_active:
|
||
raise AuthenticationFailed("用户不存在或已禁用")
|
||
return user
|
||
```
|
||
|
||
### 3.7 其余安全措施
|
||
|
||
| 措施 | 实现 |
|
||
|------|------|
|
||
| HTTPS 强制 | Nginx 配置 `return 301 https://` |
|
||
| CSRF 防护 | Cookie 设置 `SameSite=Strict; Secure; HttpOnly` |
|
||
| 请求限流 | SlowAPI + Redis 滑动窗口限流 |
|
||
| 敏感配置加密 | 实例密码 AES-256-CBC 加密存储 |
|
||
| 会话超时 | JWT 30 分钟过期 + Refresh Token 7 天 |
|
||
| 登录失败锁定 | 5 次失败锁定 15 分钟 |
|
||
| SQL 超时 | 所有查询强制 30 秒超时 |
|
||
| 审计日志防篡改 | 审计日志写入只读存储 / WORM 模式 |
|
||
|
||
---
|
||
|
||
## 4. 用户体验与产品细节优化
|
||
|
||
### 4.1 SQL 编辑器选型
|
||
|
||
**推荐: Monaco Editor (VS Code 内核)**
|
||
|
||
```typescript
|
||
// 安装
|
||
// npm install monaco-editor sql-formatter
|
||
|
||
// 核心配置
|
||
const editor = monaco.editor.create(container, {
|
||
language: 'mysql', // mysql / pgsql / redis
|
||
theme: 'vs-dark',
|
||
fontSize: 13,
|
||
fontFamily: 'JetBrains Mono, monospace',
|
||
minimap: { enabled: false },
|
||
automaticLayout: true,
|
||
scrollbar: { vertical: 'auto' },
|
||
suggestOnTriggerCharacters: true,
|
||
quickSuggestions: { other: true, comments: true, strings: true },
|
||
wordBasedSuggestions: "allDocuments",
|
||
formatOnPaste: true,
|
||
formatOnType: true,
|
||
});
|
||
|
||
// 数据库 schema 自动补全
|
||
const provideCompletionItems = async (model, position) => {
|
||
const word = model.getWordUntilPosition(position);
|
||
const tables = await fetchTables(selectedInstance.value);
|
||
return {
|
||
suggestions: tables.flatMap(t => [
|
||
{ label: t.name, kind: monaco.languages.CompletionItemKind.Class, insertText: t.name },
|
||
...t.columns.map(c => ({
|
||
label: c.name, kind: monaco.languages.CompletionItemKind.Field,
|
||
insertText: c.name, detail: c.type,
|
||
})),
|
||
]),
|
||
};
|
||
};
|
||
|
||
monaco.languages.registerCompletionItemProvider('mysql', {
|
||
triggerCharacters: ['.', ' '],
|
||
provideCompletionItems
|
||
});
|
||
```
|
||
|
||
**四数据库语法支持**:
|
||
|
||
| 数据库 | Monaco Language | 语法高亮 | 自动补全 |
|
||
|--------|----------------|---------|---------|
|
||
| MySQL | `mysql` | ✅ 内置 | 表名/字段名 |
|
||
| PostgreSQL | `pgsql` | ✅ 内置 | 表名/字段名 |
|
||
| MongoDB | `javascript` | ⚠️ 需扩展 | Collection/Field |
|
||
| Redis | 自定义 | ⚠️ 需自定义 | Key/命令 |
|
||
|
||
### 4.2 工单审批流程可视化
|
||
|
||
```vue
|
||
<!-- frontend/src/components/workflow/FlowChart.vue -->
|
||
<template>
|
||
<div class="workflow-flow">
|
||
<div v-for="(step, i) in steps" :key="step.id" class="flow-step">
|
||
<div class="step-node" :class="step.status">
|
||
<component :is="stepIcons[step.type]" class="icon" />
|
||
<div class="info">
|
||
<div class="name">{{ step.name }}</div>
|
||
<div class="assignee">{{ step.assignees.join(' / ') }}</div>
|
||
<div v-if="step.time" class="time">{{ step.time }}</div>
|
||
</div>
|
||
</div>
|
||
<div v-if="i < steps.length - 1" class="flow-arrow" :class="{ active: step.status === 'completed' }" />
|
||
</div>
|
||
</div>
|
||
</template>
|
||
|
||
<style scoped>
|
||
.workflow-flow { display: flex; align-items: center; gap: 0; }
|
||
.flow-step { display: flex; align-items: center; }
|
||
.step-node {
|
||
padding: 12px 20px; border-radius: 8px; border: 2px solid var(--dg-border);
|
||
display: flex; align-items: center; gap: 12px; min-width: 160px;
|
||
}
|
||
.step-node.completed { border-color: #22c55e; background: rgba(34,197,94,0.1); }
|
||
.step-node.active { border-color: #3b82f6; background: rgba(59,130,246,0.1); box-shadow: 0 0 0 4px rgba(59,130,246,0.2); }
|
||
.flow-arrow { width: 40px; height: 2px; background: var(--dg-border); margin: 0 4px; position: relative; }
|
||
.flow-arrow.active { background: #22c55e; }
|
||
.flow-arrow::after { content: ''; position: absolute; right: 0; top: -4px; width: 0; height: 0; border-left: 6px solid currentColor; border-top: 5px solid transparent; border-bottom: 5px solid transparent; }
|
||
</style>
|
||
```
|
||
|
||
### 4.3 通知系统
|
||
|
||
```python
|
||
# backend/app/utils/notify.py
|
||
class NotificationManager:
|
||
CHANNELS = {
|
||
'feishu': FeishuNotifier(),
|
||
'dingtalk': DingtalkNotifier(),
|
||
'wechat': WechatNotifier(),
|
||
'telegram': TelegramNotifier(),
|
||
'email': EmailNotifier(),
|
||
}
|
||
|
||
async def send(self, event: str, payload: dict, recipients: list[int]):
|
||
"""发送通知到所有已启用的渠道"""
|
||
configs = await crud.notification_config.get_enabled(db, event=event)
|
||
|
||
tasks = []
|
||
for cfg in configs:
|
||
notifier = self.CHANNELS.get(cfg.channel)
|
||
if notifier:
|
||
tasks.append(notifier.send(cfg.webhook, self._render_template(event, payload)))
|
||
|
||
await asyncio.gather(*tasks, return_exceptions=True)
|
||
|
||
def _render_template(self, event: str, payload: dict) -> str:
|
||
templates = {
|
||
'ticket.created': '【工单创建】{title}\n申请人: {applicant}\n时间: {time}',
|
||
'ticket.approved': '【审批通过】{title}\n审批人: {approver}\n时间: {time}',
|
||
'ticket.executed': '【执行完成】{title}\n状态: {status}\n耗时: {duration}',
|
||
'ticket.failed': '【执行失败】{title}\n错误: {error}',
|
||
}
|
||
return templates.get(event, '').format(**payload)
|
||
|
||
# 飞书通知实现
|
||
class FeishuNotifier:
|
||
async def send(self, webhook: str, content: str):
|
||
async with httpx.AsyncClient() as client:
|
||
await client.post(webhook, json={
|
||
"msg_type": "text",
|
||
"content": {"text": content}
|
||
})
|
||
```
|
||
|
||
### 4.4 移动端适配与深色主题
|
||
|
||
```css
|
||
/* 移动端断点 */
|
||
@media (max-width: 768px) {
|
||
.sidebar { display: none; }
|
||
.mobile-menu-btn { display: block; }
|
||
.stats-grid { grid-template-columns: repeat(2, 1fr); }
|
||
.table-container { overflow-x: scroll; }
|
||
.modal { width: 95vw; max-height: 90vh; }
|
||
}
|
||
|
||
/* 深色主题变量 */
|
||
:root {
|
||
--dg-bg-primary: #0d1117;
|
||
--dg-bg-secondary: #161b22;
|
||
--dg-bg-tertiary: #21262d;
|
||
--dg-bg-hover: #1c2128;
|
||
--dg-border: #30363d;
|
||
--dg-border-light: #21262d;
|
||
--dg-text-primary: #e6edf3;
|
||
--dg-text-secondary: #8b949e;
|
||
--dg-text-muted: #484f58;
|
||
}
|
||
|
||
/* 浅色主题（可选） */
|
||
[data-theme="light"] {
|
||
--dg-bg-primary: #ffffff;
|
||
--dg-bg-secondary: #f6f8fa;
|
||
--dg-bg-tertiary: #eaeef2;
|
||
--dg-bg-hover: #f3f4f6;
|
||
--dg-border: #d0d7de;
|
||
--dg-border-light: #eaeef2;
|
||
--dg-text-primary: #1f2328;
|
||
--dg-text-secondary: #656d76;
|
||
--dg-text-muted: #8c959f;
|
||
}
|
||
```
|
||
|
||
---
|
||
|
||
## 5. 性能与可扩展性优化
|
||
|
||
### 5.1 异步任务队列 (Celery + Redis)
|
||
|
||
```python
|
||
# backend/app/tasks/celery_tasks.py
|
||
from celery import Celery
|
||
|
||
celery_app = Celery('dataguard', broker=settings.CELERY_BROKER_URL)
|
||
|
||
@celery_app.task(bind=True, max_retries=3, default_retry_delay=60)
|
||
def execute_ticket(self, ticket_id: int, executor_id: int):
|
||
"""异步执行工单 SQL"""
|
||
try:
|
||
ticket = sync_to_async(crud.ticket.get)(ticket_id)
|
||
|
||
# 1. 预执行检查
|
||
precheck_result = sync_to_async(pre_execute_check)(ticket)
|
||
if not precheck_result.passed:
|
||
sync_to_async(crud.ticket.update)(ticket_id, status='failed',
|
||
error_msg=precheck_result.msg)
|
||
return
|
||
|
||
# 2. 执行 SQL
|
||
start = time.time()
|
||
result = sync_to_async(execute_sql)(ticket.instance_id, ticket.sql_content)
|
||
duration = time.time() - start
|
||
|
||
# 3. 记录结果
|
||
sync_to_async(crud.ticket.update)(ticket_id,
|
||
status='success',
|
||
execute_duration=duration,
|
||
result_summary=result.summary,
|
||
)
|
||
|
||
# 4. 发送通知
|
||
sync_to_async(notify_manager.send)('ticket.executed', {
|
||
'title': ticket.title,
|
||
'status': '成功',
|
||
'duration': f'{duration:.2f}s',
|
||
}, [ticket.applicant_id])
|
||
|
||
except Exception as exc:
|
||
self.retry(exc=exc)
|
||
|
||
@celery_app.task
|
||
def auto_audit_task(ticket_id: int):
|
||
"""自动审核任务"""
|
||
engine = AutoAuditEngine()
|
||
result = sync_to_async(engine.audit)(ticket_id)
|
||
|
||
if result.passed:
|
||
sync_to_async(crud.ticket.update)(ticket_id, status='pending_review')
|
||
else:
|
||
sync_to_async(crud.ticket.update)(ticket_id,
|
||
status='auto_audit_failed',
|
||
audit_score=result.score,
|
||
audit_issues=result.issues_json,
|
||
)
|
||
|
||
@celery_app.task
|
||
def backup_before_execute(ticket_id: int):
|
||
"""执行前备份受影响数据"""
|
||
ticket = sync_to_async(crud.ticket.get)(ticket_id)
|
||
backup_table = f"_dg_backup_{ticket.target_table}_{ticket_id}_{int(time.time())}"
|
||
sync_to_async(execute_sql)(ticket.instance_id,
|
||
f"CREATE TABLE {backup_table} AS SELECT * FROM {ticket.target_table} WHERE {ticket.where_clause}")
|
||
sync_to_async(crud.ticket.update)(ticket_id, backup_table=backup_table)
|
||
```
|
||
|
||
### 5.2 大 SQL 处理与不同数据库执行引擎
|
||
|
||
```python
|
||
# backend/app/core/execution_engine.py
|
||
from abc import ABC, abstractmethod
|
||
|
||
class ExecutionEngine(ABC):
|
||
@abstractmethod
|
||
async def execute(self, instance_id: int, sql: str, timeout: int = 30) -> ExecuteResult:
|
||
pass
|
||
|
||
@abstractmethod
|
||
async def explain(self, instance_id: int, sql: str) -> ExplainResult:
|
||
pass
|
||
|
||
class MySQLEngine(ExecutionEngine):
|
||
async def execute(self, instance_id, sql, timeout=30):
|
||
engine = await pool_mgr.get_mysql_engine(instance_id)
|
||
async with engine.connect() as conn:
|
||
result = await asyncio.wait_for(conn.execute(text(sql)), timeout=timeout)
|
||
return ExecuteResult(
|
||
rows=[dict(r) for r in result.mappings()],
|
||
columns=list(result.keys()),
|
||
rowcount=result.rowcount,
|
||
)
|
||
|
||
async def explain(self, instance_id, sql):
|
||
engine = await pool_mgr.get_mysql_engine(instance_id)
|
||
async with engine.connect() as conn:
|
||
result = await conn.execute(text(f"EXPLAIN {sql}"))
|
||
return ExplainResult(rows=[dict(r) for r in result.mappings()])
|
||
|
||
class PostgreSQLEngine(ExecutionEngine):
|
||
async def execute(self, instance_id, sql, timeout=30):
|
||
# 使用 EXPLAIN ANALYZE 预估 + 实际执行
|
||
...
|
||
|
||
async def explain(self, instance_id, sql):
|
||
# PostgreSQL JSON format EXPLAIN
|
||
engine = await pool_mgr.get_pg_engine(instance_id)
|
||
async with engine.connect() as conn:
|
||
result = await conn.execute(text(f"EXPLAIN (FORMAT JSON) {sql}"))
|
||
return ExplainResult(plan=result.scalar())
|
||
|
||
class MongoEngine(ExecutionEngine):
|
||
async def execute(self, instance_id, command: str, timeout=30):
|
||
client = await pool_mgr.get_mongo_client(instance_id)
|
||
db = client.get_default_database()
|
||
# MongoDB 命令解析执行
|
||
result = await db.command(eval(command))
|
||
return ExecuteResult(raw=result)
|
||
|
||
class RedisEngine(ExecutionEngine):
|
||
async def execute(self, instance_id, command: str, timeout=30):
|
||
client = await pool_mgr.get_redis_client(instance_id)
|
||
parts = command.split()
|
||
cmd, args = parts[0], parts[1:]
|
||
result = await client.execute_command(cmd, *args)
|
||
return ExecuteResult(raw=result)
|
||
|
||
# 工厂模式
|
||
class EngineFactory:
|
||
ENGINES = {
|
||
'mysql': MySQLEngine,
|
||
'postgres': PostgreSQLEngine,
|
||
'mongo': MongoEngine,
|
||
'redis': RedisEngine,
|
||
}
|
||
|
||
@classmethod
|
||
def get(cls, db_type: str) -> ExecutionEngine:
|
||
return cls.ENGINES[db_type]()
|
||
```
|
||
|
||
### 5.3 规则引擎性能优化
|
||
|
||
```python
|
||
# 1. 规则缓存
|
||
class RuleCache:
|
||
async def get_active_rules(self, db_type: str) -> list[AuditRule]:
|
||
cache_key = f"rules:active:{db_type}"
|
||
cached = await redis.get(cache_key)
|
||
if cached:
|
||
return [AuditRule.parse_raw(r) for r in json.loads(cached)]
|
||
|
||
rules = await crud.audit_rule.get_active(db, db_type=db_type)
|
||
await redis.setex(cache_key, 300, json.dumps([r.json() for r in rules]))
|
||
return rules
|
||
|
||
async def invalidate(self, db_type: str):
|
||
await redis.delete(f"rules:active:{db_type}")
|
||
|
||
# 2. 规则编译缓存（正则预编译）
|
||
class CompiledRule:
|
||
def __init__(self, rule: AuditRule):
|
||
self.rule = rule
|
||
self._compiled = re.compile(rule.pattern, re.IGNORECASE) if rule.rule_type == 'regex' else None
|
||
|
||
def match(self, sql: str) -> bool:
|
||
if self._compiled:
|
||
return bool(self._compiled.search(sql))
|
||
return False
|
||
|
||
# 3. 短路求值：block 规则匹配后立即终止
|
||
```
|
||
|
||
### 5.4 数据库连接池管理
|
||
|
||
```python
|
||
# backend/app/core/database.py
|
||
# 连接池配置参考
|
||
|
||
# MySQL (aiomysql + SQLAlchemy)
|
||
engine = create_async_engine(
|
||
url,
|
||
pool_size=5, # 常驻连接数
|
||
max_overflow=10, # 最大溢出连接
|
||
pool_recycle=3600, # 连接回收时间（秒）
|
||
pool_pre_ping=True, # 使用前探活
|
||
echo=False,
|
||
)
|
||
|
||
# PostgreSQL (asyncpg + SQLAlchemy)
|
||
engine = create_async_engine(
|
||
url,
|
||
pool_size=5,
|
||
max_overflow=10,
|
||
pool_recycle=3600,
|
||
pool_pre_ping=True,
|
||
echo=False,
|
||
)
|
||
|
||
# MongoDB (motor)
|
||
client = AsyncIOMotorClient(
|
||
uri,
|
||
maxPoolSize=10,
|
||
minPoolSize=1,
|
||
maxIdleTimeMS=60000,
|
||
serverSelectionTimeoutMS=5000,
|
||
)
|
||
|
||
# Redis (aioredis)
|
||
redis = aioredis.from_url(
|
||
url,
|
||
max_connections=20,
|
||
decode_responses=True,
|
||
health_check_interval=30,
|
||
)
|
||
```
|
||
|
||
---
|
||
|
||
## 6. 技术实现关键点
|
||
|
||
### 6.1 FastAPI 核心代码框架
|
||
|
||
```python
|
||
# backend/app/main.py
|
||
from fastapi import FastAPI
|
||
from fastapi.middleware.cors import CORSMiddleware
|
||
from contextlib import asynccontextmanager
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(app: FastAPI):
|
||
# 启动
|
||
await pool_mgr.init()
|
||
await redis_pool.init()
|
||
await celery_app.start()
|
||
yield
|
||
# 关闭
|
||
await pool_mgr.close()
|
||
await redis_pool.close()
|
||
|
||
app = FastAPI(
|
||
title="DataGuard API",
|
||
version="2.0.0",
|
||
lifespan=lifespan,
|
||
docs_url="/docs" if settings.DEBUG else None,
|
||
)
|
||
|
||
# CORS
|
||
app.add_middleware(
|
||
CORSMiddleware,
|
||
allow_origins=["https://dataguard.company.com"],
|
||
allow_credentials=True,
|
||
allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
|
||
allow_headers=["*"],
|
||
)
|
||
|
||
# 全局异常处理
|
||
@app.exception_handler(BusinessException)
|
||
async def business_exception_handler(request, exc):
|
||
return JSONResponse(status_code=400, content={"code": 400, "message": exc.message})
|
||
|
||
@app.exception_handler(PermissionDenied)
|
||
async def permission_exception_handler(request, exc):
|
||
return JSONResponse(status_code=403, content={"code": 403, "message": exc.message})
|
||
|
||
# 注册路由
|
||
app.include_router(api_router, prefix="/api/v1")
|
||
|
||
# 健康检查
|
||
@app.get("/health")
|
||
async def health_check():
|
||
return {"status": "ok", "version": "2.0.0"}
|
||
```
|
||
|
||
### 6.2 Vue3 Composition API 最佳实践
|
||
|
||
```typescript
|
||
// 组合式函数示例
|
||
// composables/useTable.ts
|
||
import { ref, computed, watch } from 'vue';
|
||
import { useQuery } from '@tanstack/vue-query';
|
||
|
||
export function useTable<T>(
|
||
fetchFn: (params: any) => Promise<{ list: T[]; total: number }>,
|
||
options: { immediate?: boolean; defaultPageSize?: number } = {}
|
||
) {
|
||
const search = ref('');
|
||
const filters = ref<Record<string, any>>({});
|
||
const pagination = ref({ page: 1, pageSize: options.defaultPageSize || 10, total: 0 });
|
||
const sorter = ref<{ field: string; order: 'ascend' | 'descend' | null }>({ field: '', order: null });
|
||
|
||
const queryKey = computed(() => [
|
||
'table',
|
||
pagination.value.page,
|
||
pagination.value.pageSize,
|
||
search.value,
|
||
filters.value,
|
||
sorter.value,
|
||
]);
|
||
|
||
const { data, isLoading, error, refetch } = useQuery({
|
||
queryKey,
|
||
queryFn: () => fetchFn({
|
||
page: pagination.value.page,
|
||
pageSize: pagination.value.pageSize,
|
||
search: search.value,
|
||
filters: filters.value,
|
||
sorter: sorter.value,
|
||
}),
|
||
select: (res) => {
|
||
pagination.value.total = res.total;
|
||
return res.list;
|
||
},
|
||
});
|
||
|
||
return {
|
||
data, isLoading, error, refetch,
|
||
search, filters, pagination, sorter,
|
||
};
|
||
}
|
||
|
||
// 页面使用
|
||
const { data: tickets, isLoading, search, pagination } = useTable(
|
||
(params) => apiTicket.list(params),
|
||
{ defaultPageSize: 20 }
|
||
);
|
||
```
|
||
|
||
### 6.3 关键 Pydantic Model
|
||
|
||
```python
|
||
# backend/app/schemas/instance.py
|
||
from pydantic import BaseModel, Field, field_validator
|
||
from typing import Literal
|
||
|
||
DbType = Literal['mysql', 'postgres', 'mongo', 'redis']
|
||
|
||
class InstanceCreate(BaseModel):
|
||
name: str = Field(..., min_length=1, max_length=100)
|
||
db_type: DbType
|
||
host: str = Field(..., pattern=r'^([a-zA-Z0-9.-]+|\d+\.\d+\.\d+\.\d+)$')
|
||
port: int = Field(..., ge=1, le=65535)
|
||
username: str = Field(..., min_length=1)
|
||
password: str = Field(..., min_length=1)
|
||
database: str | None = None
|
||
env: Literal['production', 'staging', 'development'] = 'development'
|
||
description: str | None = None
|
||
|
||
@field_validator('port')
|
||
@classmethod
|
||
def validate_port(cls, v: int, info) -> int:
|
||
db_type = info.data.get('db_type')
|
||
defaults = {'mysql': 3306, 'postgres': 5432, 'mongo': 27017, 'redis': 6379}
|
||
if v != defaults.get(db_type):
|
||
pass # 允许非默认端口，仅作提示
|
||
return v
|
||
|
||
class InstanceResponse(InstanceCreate):
|
||
id: int
|
||
status: str
|
||
version: str | None
|
||
databases: int
|
||
tables: int
|
||
size: str
|
||
connections: int
|
||
qps: int
|
||
create_time: datetime
|
||
|
||
# backend/app/schemas/ticket.py
|
||
class TicketCreate(BaseModel):
|
||
title: str = Field(..., min_length=5, max_length=200)
|
||
db_type: DbType
|
||
instance_id: int
|
||
target_database: str
|
||
sql_content: str = Field(..., min_length=1)
|
||
rollback_sql: str | None = None
|
||
reason: str = Field(..., min_length=10)
|
||
priority: Literal['high', 'medium', 'low'] = 'medium'
|
||
|
||
class TicketApprove(BaseModel):
|
||
action: Literal['approve', 'reject']
|
||
comment: str = Field(..., min_length=1, max_length=500)
|
||
|
||
class TicketExecute(BaseModel):
|
||
dry_run: bool = True # 默认先模拟执行
|
||
```
|
||
|
||
### 6.4 Docker + 多数据库驱动配置
|
||
|
||
```dockerfile
|
||
# Dockerfile
|
||
FROM python:3.11-slim
|
||
|
||
WORKDIR /app
|
||
|
||
# 安装系统依赖（MySQL/PostgreSQL 客户端 + MongoDB 驱动 + Redis）
|
||
RUN apt-get update && apt-get install -y \
|
||
gcc \
|
||
libmariadb-dev \
|
||
libpq-dev \
|
||
&& rm -rf /var/lib/apt/lists/*
|
||
|
||
COPY pyproject.toml .
|
||
RUN pip install --no-cache-dir -e ".[all]"
|
||
|
||
COPY app/ ./app/
|
||
COPY migrations/ ./migrations/
|
||
|
||
CMD ["gunicorn", "app.main:app", "-k", "uvicorn.workers.UvicornWorker", "-w", "4", "-b", "0.0.0.0:8000"]
|
||
```
|
||
|
||
```yaml
|
||
# docker-compose.yml
|
||
version: '3.8'
|
||
|
||
services:
|
||
app:
|
||
build: .
|
||
ports:
|
||
- "8000:8000"
|
||
environment:
|
||
- ENV=production
|
||
- SECRET_KEY=${SECRET_KEY}
|
||
- MYSQL_HOST=mysql
|
||
- POSTGRES_HOST=postgres
|
||
- MONGO_URI=mongodb://mongo:27017
|
||
- REDIS_URL=redis://redis:6379/0
|
||
- CELERY_BROKER_URL=redis://redis:6379/1
|
||
depends_on:
|
||
- mysql
|
||
- postgres
|
||
- mongo
|
||
- redis
|
||
volumes:
|
||
- ./logs:/app/logs
|
||
|
||
worker:
|
||
build: .
|
||
command: celery -A app.tasks.celery_tasks worker -l info -c 4
|
||
environment:
|
||
- ENV=production
|
||
- CELERY_BROKER_URL=redis://redis:6379/1
|
||
depends_on:
|
||
- redis
|
||
- mysql
|
||
- postgres
|
||
|
||
scheduler:
|
||
build: .
|
||
command: celery -A app.tasks.celery_tasks beat -l info
|
||
depends_on:
|
||
- redis
|
||
|
||
mysql:
|
||
image: mysql:8.0
|
||
environment:
|
||
MYSQL_ROOT_PASSWORD: rootpass
|
||
MYSQL_DATABASE: dataguard
|
||
volumes:
|
||
- mysql_data:/var/lib/mysql
|
||
command: --binlog-format=ROW --server-id=1
|
||
|
||
postgres:
|
||
image: postgres:16
|
||
environment:
|
||
POSTGRES_PASSWORD: pgpass
|
||
POSTGRES_DB: dataguard
|
||
volumes:
|
||
- postgres_data:/var/lib/postgresql/data
|
||
|
||
mongo:
|
||
image: mongo:7
|
||
volumes:
|
||
- mongo_data:/data/db
|
||
|
||
redis:
|
||
image: redis:7-alpine
|
||
volumes:
|
||
- redis_data:/data
|
||
|
||
nginx:
|
||
image: nginx:alpine
|
||
ports:
|
||
- "80:80"
|
||
- "443:443"
|
||
volumes:
|
||
- ./nginx.conf:/etc/nginx/nginx.conf
|
||
- ./ssl:/etc/nginx/ssl
|
||
depends_on:
|
||
- app
|
||
|
||
volumes:
|
||
mysql_data:
|
||
postgres_data:
|
||
mongo_data:
|
||
redis_data:
|
||
```
|
||
|
||
---
|
||
|
||
## 7. 部署与 DevOps
|
||
|
||
### 7.1 生产部署方案
|
||
|
||
```nginx
|
||
# nginx.conf
|
||
upstream backend {
|
||
server app:8000 max_fails=3 fail_timeout=30s;
|
||
}
|
||
|
||
server {
|
||
listen 80;
|
||
server_name dataguard.company.com;
|
||
return 301 https://$server_name$request_uri;
|
||
}
|
||
|
||
server {
|
||
listen 443 ssl http2;
|
||
server_name dataguard.company.com;
|
||
|
||
ssl_certificate /etc/nginx/ssl/cert.pem;
|
||
ssl_certificate_key /etc/nginx/ssl/key.pem;
|
||
ssl_protocols TLSv1.2 TLSv1.3;
|
||
|
||
# 前端静态文件
|
||
location / {
|
||
root /var/www/frontend;
|
||
try_files $uri $uri/ /index.html;
|
||
expires 7d;
|
||
}
|
||
|
||
# API 反向代理
|
||
location /api/ {
|
||
proxy_pass http://backend;
|
||
proxy_set_header Host $host;
|
||
proxy_set_header X-Real-IP $remote_addr;
|
||
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||
proxy_set_header X-Forwarded-Proto $scheme;
|
||
proxy_connect_timeout 30s;
|
||
proxy_send_timeout 30s;
|
||
proxy_read_timeout 30s;
|
||
}
|
||
|
||
# WebSocket（实时通知）
|
||
location /ws/ {
|
||
proxy_pass http://backend;
|
||
proxy_http_version 1.1;
|
||
proxy_set_header Upgrade $http_upgrade;
|
||
proxy_set_header Connection "upgrade";
|
||
}
|
||
}
|
||
```
|
||
|
||
```python
|
||
# Gunicorn 配置 gunicorn.conf.py
|
||
import multiprocessing
|
||
|
||
bind = "0.0.0.0:8000"
|
||
workers = multiprocessing.cpu_count() * 2 + 1
|
||
worker_class = "uvicorn.workers.UvicornWorker"
|
||
timeout = 60
|
||
keepalive = 5
|
||
errorlog = "/app/logs/gunicorn-error.log"
|
||
accesslog = "/app/logs/gunicorn-access.log"
|
||
```
|
||
|
||
### 7.2 CI/CD 流程
|
||
|
||
```yaml
|
||
# .github/workflows/ci.yml
|
||
name: CI/CD
|
||
|
||
on:
|
||
push:
|
||
branches: [main, develop]
|
||
pull_request:
|
||
branches: [main]
|
||
|
||
jobs:
|
||
test-backend:
|
||
runs-on: ubuntu-latest
|
||
services:
|
||
mysql:
|
||
image: mysql:8.0
|
||
env:
|
||
MYSQL_ROOT_PASSWORD: test
|
||
options: --health-cmd="mysqladmin ping" --health-interval=10s
|
||
redis:
|
||
image: redis:7-alpine
|
||
steps:
|
||
- uses: actions/checkout@v4
|
||
- uses: actions/setup-python@v5
|
||
with:
|
||
python-version: '3.11'
|
||
- run: pip install -e ".[dev]"
|
||
- run: pytest tests/ --cov=app --cov-report=xml
|
||
- uses: codecov/codecov-action@v3
|
||
|
||
test-frontend:
|
||
runs-on: ubuntu-latest
|
||
steps:
|
||
- uses: actions/checkout@v4
|
||
- uses: actions/setup-node@v4
|
||
with:
|
||
node-version: '20'
|
||
- run: npm ci
|
||
- run: npm run lint
|
||
- run: npm run test:unit -- --coverage
|
||
- run: npm run build
|
||
|
||
deploy:
|
||
needs: [test-backend, test-frontend]
|
||
if: github.ref == 'refs/heads/main'
|
||
runs-on: ubuntu-latest
|
||
steps:
|
||
- uses: actions/checkout@v4
|
||
- name: Build & Push Docker
|
||
run: |
|
||
docker build -t dataguard:${{ github.sha }} .
|
||
docker push dataguard:${{ github.sha }}
|
||
- name: Deploy to Server
|
||
run: |
|
||
ssh deploy@server "cd /opt/dataguard && \
|
||
docker pull dataguard:${{ github.sha }} && \
|
||
docker-compose up -d"
|
||
```
|
||
|
||
### 7.3 监控与日志
|
||
|
||
```python
|
||
# Prometheus 指标
|
||
from prometheus_client import Counter, Histogram, Gauge
|
||
|
||
request_count = Counter('dg_request_total', 'Total requests', ['method', 'endpoint', 'status'])
|
||
request_latency = Histogram('dg_request_duration_seconds', 'Request latency')
|
||
active_connections = Gauge('dg_db_active_connections', 'Active DB connections', ['db_type', 'instance'])
|
||
ticket_count = Gauge('dg_tickets_total', 'Total tickets', ['status'])
|
||
|
||
# 中间件自动收集
|
||
@app.middleware("http")
|
||
async def metrics_middleware(request: Request, call_next):
|
||
start = time.time()
|
||
response = await call_next(request)
|
||
duration = time.time() - start
|
||
|
||
request_count.labels(
|
||
method=request.method,
|
||
endpoint=request.url.path,
|
||
status=response.status_code,
|
||
).inc()
|
||
request_latency.observe(duration)
|
||
|
||
return response
|
||
```
|
||
|
||
```yaml
|
||
# prometheus.yml
|
||
scrape_configs:
|
||
- job_name: 'dataguard'
|
||
static_configs:
|
||
- targets: ['app:8000']
|
||
metrics_path: /metrics
|
||
scrape_interval: 15s
|
||
|
||
# Grafana Dashboard 关键面板:
|
||
# - API QPS / 延迟 P99
|
||
# - 数据库连接池使用率
|
||
# - 工单状态分布
|
||
# - SQL 查询 TOP10
|
||
# - 错误率告警
|
||
```
|
||
|
||
---
|
||
|
||
## 8. 迭代路线图
|
||
|
||
### 8.1 MVP（1 个月内可上线）P0
|
||
|
||
| 功能 | 优先级 | 预计耗时 | 说明 |
|
||
|------|--------|---------|------|
|
||
| 用户登录 + JWT 认证 | P0 | 3天 | FastAPI OAuth2 + Vue3 登录页 |
|
||
| MySQL 数据源管理 | P0 | 3天 | CRUD + 连接测试 + 密码加密 |
|
||
| SQL 工单提交 + 审批 | P0 | 5天 | 状态机 + 单级审批 |
|
||
| 在线 SQL 查询 | P0 | 4天 | MySQL 查询 + 结果展示 + LIMIT 限制 |
|
||
| 基础 RBAC 权限 | P0 | 3天 | admin/dba/developer 角色 |
|
||
| 工单列表 + 筛选 | P0 | 2天 | 表格 + 分页 + 状态筛选 |
|
||
|
||
**MVP 总耗时: ~20天**
|
||
|
||
### 8.2 v1.0（3 个月）P1
|
||
|
||
| 功能 | 优先级 | 预计耗时 | 说明 |
|
||
|------|--------|---------|------|
|
||
| PostgreSQL 支持 | P1 | 3天 | 数据源 + 查询引擎适配 |
|
||
| MongoDB 支持 | P1 | 5天 | 数据源 + 查询引擎 + 命令解析 |
|
||
| Redis 支持 | P1 | 4天 | 数据源 + Key 管理 + 命令执行 |
|
||
| 多级审批流程配置 | P1 | 5天 | 流程模板 + 可视化配置 |
|
||
| 审核规则引擎 | P1 | 5天 | 10条内置规则 + 自定义规则 |
|
||
| 回滚 SQL 生成 | P1 | 5天 | MySQL Binlog 逆向 |
|
||
| 数据脱敏 | P1 | 3天 | 敏感字段自动识别 + 脱敏 |
|
||
| 通知系统 | P1 | 3天 | 飞书/钉钉/邮件 |
|
||
| SQL 编辑器升级 | P1 | 4天 | Monaco Editor + 语法高亮 |
|
||
| 仪表盘 | P1 | 4天 | 统计图表 + 趋势 |
|
||
| 审计日志 | P1 | 4天 | 操作记录 + 变更对比 |
|
||
| LDAP 集成 | P1 | 3天 | 域账号登录 |
|
||
| 配置平台 | P1 | 4天 | 全局配置 + 工单规则 + DDL 拦截 |
|
||
|
||
**v1.0 总耗时: ~60天（在 MVP 基础上增量）**
|
||
|
||
### 8.3 v2.0（6 个月）P2
|
||
|
||
| 功能 | 优先级 | 预计耗时 | 说明 |
|
||
|------|--------|---------|------|
|
||
| Celery 异步任务 | P2 | 5天 | 工单异步执行 + 定时任务 |
|
||
| 自动任务调度 | P2 | 4天 | Cron 表达式 + 任务管理 |
|
||
| 数据字典自动采集 | P2 | 4天 | information_schema 同步 |
|
||
| 行级/字段级权限 | P2 | 5天 | 细粒度数据权限控制 |
|
||
| 大 SQL 处理优化 | P2 | 4天 | 分批执行 + 流式结果 |
|
||
| 查询性能分析 | P2 | 3天 | EXPLAIN 解析 + 索引建议 |
|
||
| 移动端适配 | P2 | 5天 | 响应式布局 + 触摸优化 |
|
||
| 深色/浅色主题 | P2 | 3天 | CSS 变量切换 |
|
||
| 多租户架构 | P2 | 7天 | tenant_id 隔离 |
|
||
| 数据库对比 | P2 | 5天 | 结构对比 + 数据对比 |
|
||
| 备份管理 | P2 | 5天 | 定时备份 + 恢复 |
|
||
| Prometheus + Grafana | P2 | 4天 | 指标采集 + Dashboard |
|
||
| 操作审计防篡改 | P2 | 4天 | WORM 存储 |
|
||
|
||
**v2.0 总耗时: ~60天（在 v1.0 基础上增量）**
|
||
|
||
---
|
||
|
||
## 9. 差异化竞争力与风险提醒
|
||
|
||
### 9.1 与竞品核心差异点
|
||
|
||
| 维度 | DataGuard | Bytebase | Yearning | Archery |
|
||
|------|-----------|----------|----------|---------|
|
||
| **多数据库支持** | MySQL/PG/Mongo/Redis | MySQL/PG/Mongo/Spanner/Oracle | MySQL only | MySQL/Redis/PG |
|
||
| **架构** | FastAPI + Vue3 | Go + Vue3 | Go + Vue | Django + React |
|
||
| **审批流程** | 可视化配置 + 多级 + 会签/或签 | 基础审批流 | 单级审批 | 多级审批 |
|
||
| **回滚机制** | Binlog ROW + AST + 备份 | pg_dump + 版本管理 | Binlog 逆向 | Binlog 逆向 |
|
||
| **规则引擎** | 自定义 + 版本控制 | 内置规则 | 内置规则 | 内置规则 |
|
||
| **数据脱敏** | 自动识别 + 实时脱敏 | 字段级脱敏 | 无 | 基础脱敏 |
|
||
| **Redis 管理** | Key 浏览 + 数据结构 + 序列化 | 无 | 无 | 基础 Key 操作 |
|
||
| **开源协议** | Apache-2.0 | Apache-2.0 | GPL-3.0 | Apache-2.0 |
|
||
|
||
**核心差异化**:
|
||
1. **四数据库深度支持**: 不像 Yearning 只支持 MySQL，也不像 Archery Redis 支持薄弱
|
||
2. **可视化审批流配置**: 拖拽式配置审批流程，支持会签/或签，竞品多为硬编码
|
||
3. **Redis 专业化管理**: Key 树形浏览、数据结构支持、自动序列化识别
|
||
4. **规则引擎版本控制**: 规则变更可追溯，支持自定义 Python 规则
|
||
|
||
### 9.2 潜在风险及规避建议
|
||
|
||
| 风险 | 级别 | 规避建议 |
|
||
|------|------|---------|
|
||
| **Binlog 回滚限制** | 高 | DDL 操作无法完整回滚数据，需明确告知用户；生产环境强制备份 |
|
||
| **MongoDB 事务支持** | 中 | MongoDB 4.0+ 支持多文档事务，需检测版本并给出警告 |
|
||
| **Redis 命令安全性** | 高 | 禁止 FLUSHALL/FLUSHDB/CONFIG/DEBUG 等危险命令 |
|
||
| **SQL 超时导致连接泄漏** | 高 | 所有查询强制 asyncio.wait_for 超时，连接池设置 pool_recycle |
|
||
| **敏感数据脱敏误伤** | 中 | 脱敏规则可配置白名单，支持按字段名/正则/数据类型匹配 |
|
||
| **JWT Token 泄露** | 高 | 短有效期(30min) + Refresh Token + 撤销列表；HTTPS 强制 |
|
||
| **数据库连接密码泄露** | 高 | AES 加密存储；禁止前端返回密码；使用 vault 管理密钥 |
|
||
| **Celery 任务堆积** | 中 | 监控队列长度，设置任务过期时间；Flower 实时监控 |
|
||
| **并发执行冲突** | 中 | 同一表加分布式锁（Redis SETNX），防止同时执行变更 |
|
||
| **审计日志存储膨胀** | 低 | 自动归档 90 天前日志到对象存储；热数据保留本地 |
|
||
|
||
---
|
||
|
||
*文档完成。如需调整细节或补充特定章节，请告知。*
|