# 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 ``` **后端查询执行**: ```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 ``` ### 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( fetchFn: (params: any) => Promise<{ list: T[]; total: number }>, options: { immediate?: boolean; defaultPageSize?: number } = {} ) { const search = ref(''); const filters = ref>({}); 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 天前日志到对象存储；热数据保留本地 | --- *文档完成。如需调整细节或补充特定章节，请告知。*