logan
4c9427e546
- P0: 实例访问控制全线接入权限组 (829测试) - P1: permissions字段激活 + 操作级权限检查 (843测试) - 前端: 删除数据权限页面, 权限组合并为唯一入口 - 新增: CodeMirror编辑器 + 查询模板 + 慢查询分析 - 新增: MongoDB管理页面 + Dashboard增强 - 修复: Pydantic V2迁移 + 前端TS错误修复
257 lines
9.8 KiB
Python
257 lines
9.8 KiB
Python
"""MongoDB 管理 API endpoints.
|
|
|
|
提供数据库/集合列表、集合统计、只读查询等功能。
|
|
所有端点要求进行身份认证，部分端点限制为 admin/dba 角色。
|
|
"""
|
|
|
|
import json
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import get_current_user, require_role
|
|
from app.core.permissions import check_instance_access
|
|
from app.models.instance import Instance
|
|
from app.models.user import User
|
|
from app.services.mongo_connector import MongoConnector, validate_operation
|
|
|
|
router = APIRouter(tags=["MongoDB管理"])
|
|
|
|
|
|
# ── 辅助函数 ─────────────────────────────────────────────────────
|
|
|
|
|
|
async def _get_mongo_instance(instance_id: int, db: AsyncSession) -> Instance:
|
|
"""获取并验证 MongoDB 实例。
|
|
|
|
Args:
|
|
instance_id: 实例 ID。
|
|
db: 数据库会话。
|
|
|
|
Returns:
|
|
Instance 模型对象（db_type 为 "mongo" 或 "mongodb"）。
|
|
|
|
Raises:
|
|
HTTPException: 实例不存在或不是 MongoDB 类型时。
|
|
"""
|
|
result = await db.execute(select(Instance).where(Instance.id == instance_id))
|
|
instance = result.scalar_one_or_none()
|
|
if not instance:
|
|
raise HTTPException(status_code=404, detail="数据源不存在")
|
|
if instance.db_type not in ("mongo", "mongodb"):
|
|
raise HTTPException(status_code=400, detail="该实例不是 MongoDB 类型")
|
|
return instance
|
|
|
|
|
|
# ── 请求/响应模型 ────────────────────────────────────────────────
|
|
|
|
|
|
class MongoQueryRequest(BaseModel):
|
|
"""MongoDB 只读查询请求体。"""
|
|
|
|
database: Optional[str] = Field(None, description="目标数据库名称，为空时使用实例默认库")
|
|
collection: str = Field(..., description="目标集合名称")
|
|
operation: str = Field("find", description="操作类型: find/findOne/countDocuments/distinct/aggregate")
|
|
filter: Dict[str, Any] = Field(default_factory=dict, description="查询过滤器")
|
|
projection: Optional[Dict[str, Any]] = Field(None, description="字段投影")
|
|
sort: Optional[Dict[str, int]] = Field(None, description="排序规则，如 {\"age\": -1}")
|
|
limit: int = Field(1000, ge=1, le=10000, description="返回上限")
|
|
skip: int = Field(0, ge=0, description="跳过数量")
|
|
pipeline: Optional[List[Dict[str, Any]]] = Field(None, description="aggregate 管道")
|
|
field: Optional[str] = Field(None, description="distinct 字段名")
|
|
|
|
|
|
class MongoChangeRequest(BaseModel):
|
|
"""MongoDB 变更操作请求体（工单执行用）。"""
|
|
|
|
database: Optional[str] = Field(None, description="目标数据库名称")
|
|
collection: str = Field(..., description="目标集合名称")
|
|
operation: str = Field(..., description="操作类型: insertOne/insertMany/updateOne/updateMany/deleteOne/deleteMany")
|
|
document: Optional[Dict[str, Any]] = Field(None, description="insertOne 文档")
|
|
documents: Optional[List[Dict[str, Any]]] = Field(None, description="insertMany 文档列表")
|
|
update: Optional[Dict[str, Any]] = Field(None, description="update 操作文档")
|
|
replacement: Optional[Dict[str, Any]] = Field(None, description="replaceOne 替换文档")
|
|
filter: Dict[str, Any] = Field(default_factory=dict, description="查询过滤器")
|
|
|
|
|
|
# ── Endpoints ────────────────────────────────────────────────────
|
|
|
|
|
|
@router.get("/mongo/{instance_id}/databases")
|
|
async def list_mongo_databases(
|
|
instance_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""列出 MongoDB 实例中的所有数据库。"""
|
|
instance = await _get_mongo_instance(instance_id, db)
|
|
# P0: 权限检查
|
|
await check_instance_access(current_user, instance, db)
|
|
try:
|
|
databases = await MongoConnector.list_databases(instance)
|
|
return {"databases": databases}
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"获取数据库列表失败: {exc}")
|
|
|
|
|
|
@router.get("/mongo/{instance_id}/databases/{database}/collections")
|
|
async def list_mongo_collections(
|
|
instance_id: int,
|
|
database: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""列出指定数据库中的所有集合。"""
|
|
instance = await _get_mongo_instance(instance_id, db)
|
|
# P0: 权限检查
|
|
await check_instance_access(current_user, instance, db)
|
|
try:
|
|
collections = await MongoConnector.list_collections(instance, database)
|
|
return {"database": database, "collections": collections}
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"获取集合列表失败: {exc}")
|
|
|
|
|
|
@router.get("/mongo/{instance_id}/collections/{collection}/stats")
|
|
async def get_mongo_collection_stats(
|
|
instance_id: int,
|
|
collection: str,
|
|
database: Optional[str] = Query(None, description="数据库名称"),
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""获取指定集合的统计信息。"""
|
|
instance = await _get_mongo_instance(instance_id, db)
|
|
# P0: 权限检查
|
|
await check_instance_access(current_user, instance, db)
|
|
db_name = database or getattr(instance, "database", None)
|
|
if not db_name:
|
|
raise HTTPException(status_code=400, detail="未指定数据库名称")
|
|
try:
|
|
stats = await MongoConnector.get_collection_stats(instance, db_name, collection)
|
|
return stats
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"获取集合统计失败: {exc}")
|
|
|
|
|
|
@router.post("/mongo/{instance_id}/query")
|
|
async def execute_mongo_query(
|
|
instance_id: int,
|
|
request: MongoQueryRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""执行 MongoDB 只读查询。
|
|
|
|
支持的只读操作:
|
|
- find: 需 filter, projection, sort, limit, skip
|
|
- findOne: 需 filter, projection
|
|
- countDocuments: 需 filter
|
|
- estimatedDocumentCount: 无需额外参数
|
|
- distinct: 需 field, filter
|
|
- aggregate: 需 pipeline
|
|
|
|
请求示例:
|
|
```json
|
|
{
|
|
"database": "test",
|
|
"collection": "users",
|
|
"operation": "find",
|
|
"filter": {"age": {"$gt": 18}},
|
|
"projection": {"name": 1, "age": 1},
|
|
"sort": {"age": -1},
|
|
"limit": 100
|
|
}
|
|
```
|
|
"""
|
|
instance = await _get_mongo_instance(instance_id, db)
|
|
# P0: 权限检查
|
|
await check_instance_access(current_user, instance, db)
|
|
db_name = request.database or getattr(instance, "database", None)
|
|
if not db_name:
|
|
raise HTTPException(status_code=400, detail="未指定数据库名称")
|
|
|
|
# 验证只读操作
|
|
is_readonly, error_msg = validate_operation(request.operation)
|
|
if not is_readonly:
|
|
raise HTTPException(status_code=403, detail=error_msg)
|
|
|
|
try:
|
|
query_json = request.model_dump(exclude_none=True, exclude={"database"})
|
|
# 移除未设置的 pipeline/field 避免错误分发
|
|
if request.operation != "aggregate":
|
|
query_json.pop("pipeline", None)
|
|
if request.operation != "distinct":
|
|
query_json.pop("field", None)
|
|
|
|
result = await MongoConnector.execute_query_json(
|
|
instance, db_name, query_json
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"operation": request.operation,
|
|
"collection": request.collection,
|
|
"result": result,
|
|
}
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"查询执行失败: {exc}")
|
|
|
|
|
|
@router.post("/mongo/{instance_id}/execute")
|
|
async def execute_mongo_change(
|
|
instance_id: int,
|
|
request: MongoChangeRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin", "dba")),
|
|
):
|
|
"""执行 MongoDB 变更操作（仅 admin/dba）。
|
|
|
|
用于工单变更执行，支持 insert/update/delete 等写操作。
|
|
"""
|
|
instance = await _get_mongo_instance(instance_id, db)
|
|
# P0: 权限检查
|
|
await check_instance_access(current_user, instance, db)
|
|
db_name = request.database or getattr(instance, "database", None)
|
|
if not db_name:
|
|
raise HTTPException(status_code=400, detail="未指定数据库名称")
|
|
|
|
try:
|
|
change_json = request.model_dump(exclude_none=True, exclude={"database"})
|
|
result = await MongoConnector.execute_change_json(
|
|
instance, db_name, change_json
|
|
)
|
|
return {
|
|
"success": True,
|
|
"operation": request.operation,
|
|
"collection": request.collection,
|
|
"result": result,
|
|
}
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"变更执行失败: {exc}")
|
|
|
|
|
|
@router.post("/mongo/{instance_id}/test-connection")
|
|
async def test_mongo_connection(
|
|
instance_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""测试 MongoDB 实例连接。"""
|
|
instance = await _get_mongo_instance(instance_id, db)
|
|
# P0: 权限检查
|
|
await check_instance_access(current_user, instance, db)
|
|
try:
|
|
result = await MongoConnector.test_connection(instance)
|
|
return result.to_dict()
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=f"连接测试失败: {exc}")
|