0 Fork 0

Files

T

loganandClaude Opus 4.7 439155c92e Initial commit: SQL audit platform v0.1.0

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>

2026-05-22 11:11:57 +08:00

67 KiB

Raw Blame History

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): 服务端状态管理，自动缓存/刷新/去重

// 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 自动生成前端类型:

# 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

前端类型生成命令:

# 使用 openapi-typescript 从 FastAPI OpenAPI 文档自动生成
npx openapi-typescript http://localhost:8000/openapi.json --output src/types/api.ts

1.4 多环境隔离与多租户

# 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 过滤

# 全局租户上下文
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 加密存储
连接复用	查询审计复用只读连接，工单执行使用独立连接

代码结构:

# 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逆向 + 备份恢复)

状态机设计:

# 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}")
        ...

自动审核引擎:

# 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):

# 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 编辑，无语法高亮、无自动补全。

优化方案:

// 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>

后端查询执行:

# 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 读取）还是手动维护。

# 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 / 压缩数据

# 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
数据源健康状态
待办审批提醒

# 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 审核规则引擎

当前不足: 规则硬编码，不支持自定义和版本控制。

优化方案: 规则引擎支持自定义 + 版本控制。

# 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)

规则引擎执行:

# 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 注入防护

# 后端：严格使用参数化查询，禁止字符串拼接 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 限制命令集	按需配置可执行命令

# 执行前权限校验
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 敏感数据脱敏

# 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 操作审计日志（含变更前后对比）

# 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 + 行级/字段级数据权限

# 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 最佳实践

# 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 内核)

// 安装
// 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 工单审批流程可视化

<!-- 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 通知系统

# 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 移动端适配与深色主题

/* 移动端断点 */
@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)

# 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 处理与不同数据库执行引擎

# 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 规则引擎性能优化

# 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 数据库连接池管理

# 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 核心代码框架

# 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 最佳实践

// 组合式函数示例
// 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

# 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
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"]

# 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.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";
    }
}

# 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 流程

# .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 监控与日志

# 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

# 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

核心差异化:

四数据库深度支持: 不像 Yearning 只支持 MySQL，也不像 Archery Redis 支持薄弱
可视化审批流配置: 拖拽式配置审批流程，支持会签/或签，竞品多为硬编码
Redis 专业化管理: Key 树形浏览、数据结构支持、自动序列化识别
规则引擎版本控制: 规则变更可追溯，支持自定义 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 天前日志到对象存储；热数据保留本地

文档完成。如需调整细节或补充特定章节，请告知。

67 KiB Raw Blame History Unescape Escape

DataGuard 数据库变更审核平台 — 全面系统性优化指南

1. 总体架构优化建议

1.1 FastAPI 后端最佳项目结构

1.2 Vue3 前端架构优化

1.3 前后端类型同步方案

1.4 多环境隔离与多租户

2. 核心功能模块深度优化

2.1 数据源管理

2.2 SQL 工单全生命周期

2.3 在线 SQL 查询器

2.4 数据字典

2.5 Redis 专用管理功能

2.6 仪表盘与监控

2.7 审核规则引擎

3. 安全与合规强化

3.1 SQL 注入防护

3.2 不同数据库的权限最小化

3.3 敏感数据脱敏

3.4 操作审计日志（含变更前后对比）

3.5 RBAC + 行级/字段级数据权限

3.6 JWT + OAuth2 最佳实践

3.7 其余安全措施

4. 用户体验与产品细节优化

4.1 SQL 编辑器选型

4.2 工单审批流程可视化

4.3 通知系统

4.4 移动端适配与深色主题

5. 性能与可扩展性优化

5.1 异步任务队列 (Celery + Redis)

5.2 大 SQL 处理与不同数据库执行引擎

5.3 规则引擎性能优化

5.4 数据库连接池管理

6. 技术实现关键点

6.1 FastAPI 核心代码框架

6.2 Vue3 Composition API 最佳实践

6.3 关键 Pydantic Model

6.4 Docker + 多数据库驱动配置

7. 部署与 DevOps

7.1 生产部署方案

7.2 CI/CD 流程

7.3 监控与日志

8. 迭代路线图

8.1 MVP（1 个月内可上线）P0

8.2 v1.0（3 个月）P1

8.3 v2.0（6 个月）P2

9. 差异化竞争力与风险提醒

9.1 与竞品核心差异点

9.2 潜在风险及规避建议

67 KiB

Raw Blame History