880 lines
34 KiB
Python
880 lines
34 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any
|
||
|
||
from core.config import MONGO_DB_NAME, utc_now
|
||
from .mongo_metrics import build_mongo_client
|
||
|
||
|
||
# 文档浏览默认每页 100 条，和页面默认值保持一致。
|
||
DEFAULT_PAGE_SIZE = 100
|
||
# 单次最多放开到 200 条，避免把浏览器一次撑爆。
|
||
MAX_PAGE_SIZE = 200
|
||
# 浏览页默认按 _id 倒序，和大多数运维排查习惯一致。
|
||
DEFAULT_SORT_SPEC = {"_id": -1}
|
||
# 列表里的 JSON 预览文本做一个上限，避免整行被超长文档拉坏。
|
||
PREVIEW_TEXT_LIMIT = 280
|
||
# Mongo 常见索引类型统一在这里收口，页面允许直接传这些值。
|
||
ALLOWED_INDEX_DIRECTIONS = {"text", "hashed", "2dsphere", "2d"}
|
||
|
||
|
||
def require_bson_json_util() -> Any:
|
||
try:
|
||
# PyMongo 安装后会自带 bson 包，这里延迟导入避免缺依赖时整个服务起不来。
|
||
from bson import json_util
|
||
except Exception as exc: # noqa: BLE001
|
||
# 管理页不可用时直接把依赖错误返回给上层。
|
||
raise RuntimeError(f"bson unavailable: {exc}") from exc
|
||
|
||
# 返回 bson 的扩展 JSON 工具。
|
||
return json_util
|
||
|
||
|
||
def get_mongo_client() -> Any:
|
||
# 统一复用现有 Mongo 建连逻辑，避免管理页和监控页配置口径分叉。
|
||
client, error = build_mongo_client()
|
||
|
||
# 建连失败时直接抛统一错误。
|
||
if client is None:
|
||
raise RuntimeError(error)
|
||
|
||
# 先做一次 ping，尽早把连通性错误暴露出来。
|
||
client.admin.command("ping")
|
||
return client
|
||
|
||
|
||
def clamp_page_size(value: Any, default: int = DEFAULT_PAGE_SIZE) -> int:
|
||
try:
|
||
# 页面传来的值可能是字符串，这里统一转整数。
|
||
numeric = int(value)
|
||
except Exception: # noqa: BLE001
|
||
numeric = default
|
||
|
||
# 最终把页大小限制在一个安全区间内。
|
||
return max(1, min(numeric, MAX_PAGE_SIZE))
|
||
|
||
|
||
def clamp_page(value: Any, default: int = 1) -> int:
|
||
try:
|
||
# 页码同样允许前端传字符串。
|
||
numeric = int(value)
|
||
except Exception: # noqa: BLE001
|
||
numeric = default
|
||
|
||
# 页码至少从 1 开始。
|
||
return max(1, numeric)
|
||
|
||
|
||
def require_name(value: Any, field_name: str) -> str:
|
||
# 数据库名、集合名、索引名都按最基础的字符串校验处理。
|
||
name = str(value or "").strip()
|
||
|
||
# 名称缺失时直接报错。
|
||
if not name:
|
||
raise ValueError(f"{field_name} is required")
|
||
|
||
# Mongo 名称里不允许带空字节。
|
||
if "\x00" in name:
|
||
raise ValueError(f"{field_name} contains invalid character")
|
||
|
||
return name
|
||
|
||
|
||
def ensure_mutable_collection_name(name: str) -> str:
|
||
collection_name = require_name(name, "collection")
|
||
|
||
# 系统集合默认不开放结构变更，避免误删系统元数据。
|
||
if collection_name.startswith("system."):
|
||
raise ValueError("system collections are read only in this page")
|
||
|
||
return collection_name
|
||
|
||
|
||
def parse_extended_json(value: Any, field_name: str, *, default: Any = None) -> Any:
|
||
# 空值时直接回退默认值，方便页面把空输入视作 {} 或 []。
|
||
if value is None:
|
||
return default
|
||
|
||
# 字符串输入按原文解析，避免前端先 stringify 再导致转义双重嵌套。
|
||
if isinstance(value, str):
|
||
text = value.strip()
|
||
if not text:
|
||
return default
|
||
else:
|
||
# 对象输入统一先转成标准 JSON 文本，再交给 bson 扩展解析。
|
||
text = json.dumps(value, ensure_ascii=False)
|
||
|
||
try:
|
||
return require_bson_json_util().loads(text)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"invalid {field_name}: {exc}") from exc
|
||
|
||
|
||
def dump_extended_json(value: Any, *, pretty: bool = False) -> str:
|
||
helper = require_bson_json_util()
|
||
options = getattr(helper, "RELAXED_JSON_OPTIONS", None)
|
||
|
||
# 列表页和编辑器都统一输出 relaxed extended JSON，前后端回写更稳定。
|
||
return helper.dumps(
|
||
value,
|
||
ensure_ascii=False,
|
||
indent=2 if pretty else None,
|
||
json_options=options,
|
||
)
|
||
|
||
|
||
def to_builtin_json(value: Any) -> Any:
|
||
# 先转扩展 JSON，再回到纯 Python 结构，这样 http_app 可以继续用标准 json.dumps。
|
||
return json.loads(dump_extended_json(value, pretty=False))
|
||
|
||
|
||
def compact_json_preview(value: Any, limit: int = PREVIEW_TEXT_LIMIT) -> str:
|
||
# 预览统一压成单行，避免表格高度失控。
|
||
compact = dump_extended_json(value, pretty=False).replace("\n", " ").strip()
|
||
if len(compact) <= limit:
|
||
return compact
|
||
return f"{compact[:limit]}..."
|
||
|
||
|
||
def mongo_document_row(document: dict[str, Any], *, page: int, position: int) -> dict[str, Any]:
|
||
# 单条文档既返回结构化内容，也返回编辑器直接可用的 JSON 文本。
|
||
document_json = dump_extended_json(document, pretty=True)
|
||
document_id = document.get("_id")
|
||
|
||
return {
|
||
"key": f"{page}-{position}-{compact_json_preview(document_id, limit=64)}",
|
||
"idText": compact_json_preview(document_id, limit=72),
|
||
"idFilterText": dump_extended_json({"_id": document_id}, pretty=True),
|
||
"previewText": compact_json_preview(document),
|
||
"jsonText": document_json,
|
||
"document": to_builtin_json(document),
|
||
}
|
||
|
||
|
||
def normalize_sort_spec(sort_value: Any) -> list[tuple[str, int]]:
|
||
parsed = parse_extended_json(sort_value, "sort", default=DEFAULT_SORT_SPEC)
|
||
|
||
# 浏览页主路径只支持对象排序，界面也按这个格式产出。
|
||
if not isinstance(parsed, dict):
|
||
raise ValueError("sort must be a json object")
|
||
|
||
items: list[tuple[str, int]] = []
|
||
for field_name, direction in parsed.items():
|
||
name = str(field_name or "").strip()
|
||
if not name:
|
||
continue
|
||
|
||
try:
|
||
numeric_direction = int(direction)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(f"invalid sort direction for {name}") from exc
|
||
|
||
if numeric_direction not in {-1, 1}:
|
||
raise ValueError(f"sort direction for {name} must be 1 or -1")
|
||
|
||
items.append((name, numeric_direction))
|
||
|
||
# 排序对象全空时回退默认值。
|
||
return items or list(DEFAULT_SORT_SPEC.items())
|
||
|
||
|
||
def normalize_index_keys(keys_value: Any) -> list[tuple[str, Any]]:
|
||
parsed = parse_extended_json(keys_value, "keys", default={})
|
||
|
||
# 索引创建页同样只接受对象格式。
|
||
if not isinstance(parsed, dict):
|
||
raise ValueError("keys must be a json object")
|
||
|
||
keys: list[tuple[str, Any]] = []
|
||
for field_name, direction in parsed.items():
|
||
name = str(field_name or "").strip()
|
||
if not name:
|
||
continue
|
||
|
||
if isinstance(direction, str):
|
||
normalized_direction = direction.strip().lower()
|
||
if normalized_direction in ALLOWED_INDEX_DIRECTIONS:
|
||
keys.append((name, normalized_direction))
|
||
continue
|
||
|
||
try:
|
||
numeric_direction = int(normalized_direction)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(
|
||
f"index direction for {name} must be 1, -1, text, hashed, 2dsphere or 2d",
|
||
) from exc
|
||
else:
|
||
try:
|
||
numeric_direction = int(direction)
|
||
except Exception as exc: # noqa: BLE001
|
||
raise ValueError(
|
||
f"index direction for {name} must be 1, -1, text, hashed, 2dsphere or 2d",
|
||
) from exc
|
||
|
||
if numeric_direction not in {-1, 1}:
|
||
raise ValueError(f"index direction for {name} must be 1, -1, text, hashed, 2dsphere or 2d")
|
||
|
||
keys.append((name, numeric_direction))
|
||
|
||
if not keys:
|
||
raise ValueError("keys must not be empty")
|
||
|
||
return keys
|
||
|
||
|
||
def require_json_object(value: Any, field_name: str, *, default: dict[str, Any] | None = None) -> dict[str, Any]:
|
||
parsed = parse_extended_json(value, field_name, default=default)
|
||
if parsed is None:
|
||
return {}
|
||
|
||
# filter / projection / options / replacement 都要求对象。
|
||
if not isinstance(parsed, dict):
|
||
raise ValueError(f"{field_name} must be a json object")
|
||
|
||
return parsed
|
||
|
||
|
||
def require_json_array(value: Any, field_name: str, *, default: list[Any] | None = None) -> list[Any]:
|
||
parsed = parse_extended_json(value, field_name, default=default)
|
||
if parsed is None:
|
||
return []
|
||
|
||
# aggregation pipeline 明确要求数组。
|
||
if not isinstance(parsed, list):
|
||
raise ValueError(f"{field_name} must be a json array")
|
||
|
||
return parsed
|
||
|
||
|
||
def require_insert_documents(value: Any) -> tuple[list[dict[str, Any]], bool]:
|
||
# Mongo 写入既支持单文档，也支持批量数组。
|
||
parsed = parse_extended_json(value, "document", default={})
|
||
|
||
if isinstance(parsed, dict):
|
||
return [parsed], False
|
||
|
||
if not isinstance(parsed, list):
|
||
raise ValueError("document must be a json object or json array")
|
||
|
||
if not parsed:
|
||
raise ValueError("document array must not be empty")
|
||
|
||
documents: list[dict[str, Any]] = []
|
||
for index, item in enumerate(parsed, start=1):
|
||
if not isinstance(item, dict):
|
||
raise ValueError(f"document[{index}] must be a json object")
|
||
documents.append(item)
|
||
|
||
return documents, True
|
||
|
||
|
||
def ensure_safe_aggregate_pipeline(pipeline: list[Any]) -> list[Any]:
|
||
for stage in pipeline:
|
||
# 每个 stage 都必须是对象，否则驱动会直接报错，这里提前统一提示。
|
||
if not isinstance(stage, dict):
|
||
raise ValueError("aggregate pipeline stage must be a json object")
|
||
|
||
# 当前页面不开放会落地写数据的聚合阶段。
|
||
stage_names = {str(key or "") for key in stage.keys()}
|
||
if "$out" in stage_names or "$merge" in stage_names:
|
||
raise ValueError("aggregate pipeline does not allow $out or $merge")
|
||
|
||
return pipeline
|
||
|
||
|
||
def collection_handle(client: Any, database_name: str, collection_name: str) -> Any:
|
||
# 集合句柄统一从这里拿，保证数据库名和集合名都先校验。
|
||
return client[require_name(database_name, "database")][require_name(collection_name, "collection")]
|
||
|
||
|
||
def collection_stats_payload(client: Any, database_name: str, collection_name: str) -> dict[str, Any]:
|
||
database = client[require_name(database_name, "database")]
|
||
collection = require_name(collection_name, "collection")
|
||
|
||
# collStats 能拿到集合大小、文档数、索引数等管理页最常用信息。
|
||
stats = database.command("collStats", collection)
|
||
index_list = list(database[collection].list_indexes())
|
||
|
||
indexes = []
|
||
for item in index_list:
|
||
key_spec = dict(item.get("key") or {})
|
||
indexes.append(
|
||
{
|
||
"name": str(item.get("name") or "").strip(),
|
||
"keys": to_builtin_json(key_spec),
|
||
"keysText": dump_extended_json(key_spec, pretty=True),
|
||
"unique": bool(item.get("unique")),
|
||
"sparse": bool(item.get("sparse")),
|
||
"hidden": bool(item.get("hidden")),
|
||
"expireAfterSeconds": item.get("expireAfterSeconds"),
|
||
"partialFilterExpression": to_builtin_json(item.get("partialFilterExpression"))
|
||
if item.get("partialFilterExpression") is not None
|
||
else None,
|
||
"partialFilterExpressionText": dump_extended_json(item.get("partialFilterExpression"), pretty=True)
|
||
if item.get("partialFilterExpression") is not None
|
||
else "",
|
||
}
|
||
)
|
||
|
||
# 把 _id_ 放在第一行，其他索引按名称排序。
|
||
indexes.sort(key=lambda item: (0 if item["name"] == "_id_" else 1, item["name"]))
|
||
|
||
return {
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"updatedAt": utc_now(),
|
||
"stats": {
|
||
"count": int(stats.get("count") or 0),
|
||
"sizeMb": round(float(stats.get("size") or 0) / 1024 / 1024, 2),
|
||
"storageSizeMb": round(float(stats.get("storageSize") or 0) / 1024 / 1024, 2),
|
||
"avgObjSizeBytes": round(float(stats.get("avgObjSize") or 0), 2),
|
||
"indexCount": int(stats.get("nindexes") or 0),
|
||
"totalIndexSizeMb": round(float(stats.get("totalIndexSize") or 0) / 1024 / 1024, 2),
|
||
"capped": bool(stats.get("capped")),
|
||
"wiredTigerUri": str((stats.get("wiredTiger") or {}).get("uri") or "").strip(),
|
||
},
|
||
"indexes": indexes,
|
||
}
|
||
|
||
|
||
def get_mongo_admin_overview_payload() -> dict[str, Any]:
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
# 概览页既要列数据库，也要列集合，所以这里一次性把树形结构拼完。
|
||
databases_raw = list((client.admin.command("listDatabases", 1).get("databases") or []))
|
||
overview_items: list[dict[str, Any]] = []
|
||
|
||
for database_item in databases_raw:
|
||
database_name = str(database_item.get("name") or "").strip()
|
||
if not database_name:
|
||
continue
|
||
|
||
database = client[database_name]
|
||
collection_items: list[dict[str, Any]] = []
|
||
|
||
# listCollections 先拿到集合名和类型，再按需补充 stats。
|
||
for collection_info in database.list_collections():
|
||
collection_name = str(collection_info.get("name") or "").strip()
|
||
if not collection_name:
|
||
continue
|
||
|
||
try:
|
||
collection_stats = database.command("collStats", collection_name)
|
||
collection_items.append(
|
||
{
|
||
"name": collection_name,
|
||
"type": str(collection_info.get("type") or "collection").strip() or "collection",
|
||
"documentCount": int(collection_stats.get("count") or 0),
|
||
"sizeMb": round(float(collection_stats.get("size") or 0) / 1024 / 1024, 2),
|
||
"storageSizeMb": round(
|
||
float(collection_stats.get("storageSize") or 0) / 1024 / 1024,
|
||
2,
|
||
),
|
||
"indexCount": int(collection_stats.get("nindexes") or 0),
|
||
"capped": bool(collection_stats.get("capped")),
|
||
"error": "",
|
||
}
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
# 某个集合取 stats 失败时仍保留在树里，方便继续点进去排查。
|
||
collection_items.append(
|
||
{
|
||
"name": collection_name,
|
||
"type": str(collection_info.get("type") or "collection").strip() or "collection",
|
||
"documentCount": 0,
|
||
"sizeMb": 0.0,
|
||
"storageSizeMb": 0.0,
|
||
"indexCount": 0,
|
||
"capped": False,
|
||
"error": str(exc),
|
||
}
|
||
)
|
||
|
||
# 集合列表默认按名字排序，侧边 explorer 更稳定。
|
||
collection_items.sort(key=lambda item: item["name"].lower())
|
||
|
||
overview_items.append(
|
||
{
|
||
"name": database_name,
|
||
"empty": bool(database_item.get("empty")),
|
||
"sizeOnDiskMb": round(float(database_item.get("sizeOnDisk") or 0) / 1024 / 1024, 2),
|
||
"collectionCount": len(collection_items),
|
||
"collections": collection_items,
|
||
}
|
||
)
|
||
|
||
# 聚焦库优先，其余按磁盘占用倒序，和现有 Mongo 指标页一致。
|
||
overview_items.sort(
|
||
key=lambda item: (
|
||
0 if item["name"] == MONGO_DB_NAME else 1,
|
||
-float(item.get("sizeOnDiskMb") or 0.0),
|
||
item["name"].lower(),
|
||
)
|
||
)
|
||
|
||
default_database = next((item["name"] for item in overview_items if item["name"] == MONGO_DB_NAME), "")
|
||
if not default_database and overview_items:
|
||
default_database = overview_items[0]["name"]
|
||
|
||
default_collection = ""
|
||
if default_database:
|
||
default_database_item = next((item for item in overview_items if item["name"] == default_database), None)
|
||
if default_database_item and default_database_item["collections"]:
|
||
default_collection = default_database_item["collections"][0]["name"]
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"focusedDatabase": MONGO_DB_NAME,
|
||
"selectedDatabase": default_database,
|
||
"selectedCollection": default_collection,
|
||
"items": overview_items,
|
||
}
|
||
finally:
|
||
# 管理页操作完成后立即关连接，避免长期挂着无用游标。
|
||
client.close()
|
||
|
||
|
||
def get_mongo_collection_detail_payload(database_name: str, collection_name: str) -> dict[str, Any]:
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
return collection_stats_payload(client, database_name, collection_name)
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def query_mongo_documents_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = require_name(payload.get("collection"), "collection")
|
||
page = clamp_page(payload.get("page"), default=1)
|
||
page_size = clamp_page_size(payload.get("pageSize"), default=DEFAULT_PAGE_SIZE)
|
||
filter_spec = require_json_object(payload.get("filter"), "filter", default={})
|
||
projection_spec = require_json_object(payload.get("projection"), "projection", default={})
|
||
sort_spec = normalize_sort_spec(payload.get("sort"))
|
||
|
||
# 页面上的编辑和删除都依赖 _id，所以这里强制保留。
|
||
if projection_spec.get("_id") == 0:
|
||
projection_spec = dict(projection_spec)
|
||
projection_spec.pop("_id", None)
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection = collection_handle(client, database_name, collection_name)
|
||
total = collection.count_documents(filter_spec)
|
||
skip = max(page - 1, 0) * page_size
|
||
|
||
cursor = collection.find(filter_spec, projection_spec or None).sort(sort_spec).skip(skip).limit(page_size)
|
||
documents = list(cursor)
|
||
rows = [
|
||
mongo_document_row(document, page=page, position=skip + index)
|
||
for index, document in enumerate(documents, start=1)
|
||
]
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"page": page,
|
||
"pageSize": page_size,
|
||
"total": total,
|
||
"pageCount": max((total + page_size - 1) // page_size, 1),
|
||
"filterText": dump_extended_json(filter_spec, pretty=True),
|
||
"projectionText": dump_extended_json(projection_spec, pretty=True),
|
||
"sortText": dump_extended_json(dict(sort_spec), pretty=True),
|
||
"items": rows,
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def insert_mongo_document_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
documents, many = require_insert_documents(payload.get("document"))
|
||
ordered = bool(payload.get("ordered", True))
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection = collection_handle(client, database_name, collection_name)
|
||
if many:
|
||
result = collection.insert_many(documents, ordered=ordered)
|
||
inserted_ids = [to_builtin_json(inserted_id) for inserted_id in result.inserted_ids]
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"insertedCount": len(inserted_ids),
|
||
"insertedIds": inserted_ids,
|
||
"message": "insertMany completed",
|
||
}
|
||
|
||
result = collection.insert_one(documents[0])
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"insertedCount": 1,
|
||
"insertedId": to_builtin_json(result.inserted_id),
|
||
"insertedIdText": dump_extended_json(result.inserted_id, pretty=False),
|
||
"message": "insertOne completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def replace_mongo_document_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
filter_spec = require_json_object(payload.get("filter"), "filter", default={})
|
||
replacement = require_json_object(payload.get("document"), "document")
|
||
upsert = bool(payload.get("upsert"))
|
||
|
||
# replaceOne 只接受完整文档，不能混进 $set 这类更新操作符。
|
||
if any(str(key).startswith("$") for key in replacement.keys()):
|
||
raise ValueError("replace document must not contain update operators")
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
result = collection_handle(client, database_name, collection_name).replace_one(
|
||
filter_spec,
|
||
replacement,
|
||
upsert=upsert,
|
||
)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"matchedCount": int(result.matched_count),
|
||
"modifiedCount": int(result.modified_count),
|
||
"upsertedId": to_builtin_json(result.upserted_id) if result.upserted_id is not None else None,
|
||
"message": "replaceOne completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def update_mongo_documents_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
filter_spec = require_json_object(payload.get("filter"), "filter", default={})
|
||
update_spec = parse_extended_json(payload.get("update"), "update", default={})
|
||
many = bool(payload.get("many"))
|
||
upsert = bool(payload.get("upsert"))
|
||
allow_empty_filter = bool(payload.get("allowEmptyFilter"))
|
||
|
||
# 空 filter + 多文档更新风险过高，前端必须显式确认。
|
||
if many and not filter_spec and not allow_empty_filter:
|
||
raise ValueError("empty filter requires explicit confirmation")
|
||
|
||
# update 支持操作符对象或更新管道数组。
|
||
if isinstance(update_spec, dict):
|
||
if not update_spec:
|
||
raise ValueError("update must not be empty")
|
||
if not any(str(key).startswith("$") for key in update_spec.keys()):
|
||
raise ValueError("update must contain update operators or use replace")
|
||
elif isinstance(update_spec, list):
|
||
ensure_safe_aggregate_pipeline(update_spec)
|
||
else:
|
||
raise ValueError("update must be a json object or json array")
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection = collection_handle(client, database_name, collection_name)
|
||
result = (
|
||
collection.update_many(filter_spec, update_spec, upsert=upsert)
|
||
if many
|
||
else collection.update_one(filter_spec, update_spec, upsert=upsert)
|
||
)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"matchedCount": int(result.matched_count),
|
||
"modifiedCount": int(result.modified_count),
|
||
"upsertedId": to_builtin_json(result.upserted_id) if result.upserted_id is not None else None,
|
||
"message": "updateMany completed" if many else "updateOne completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def delete_mongo_documents_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
filter_spec = require_json_object(payload.get("filter"), "filter", default={})
|
||
many = bool(payload.get("many"))
|
||
allow_empty_filter = bool(payload.get("allowEmptyFilter"))
|
||
|
||
# deleteMany 空 filter 同样要求显式确认。
|
||
if many and not filter_spec and not allow_empty_filter:
|
||
raise ValueError("empty filter requires explicit confirmation")
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection = collection_handle(client, database_name, collection_name)
|
||
result = collection.delete_many(filter_spec) if many else collection.delete_one(filter_spec)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"deletedCount": int(result.deleted_count),
|
||
"message": "deleteMany completed" if many else "deleteOne completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def aggregate_mongo_documents_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = require_name(payload.get("collection"), "collection")
|
||
page = clamp_page(payload.get("page"), default=1)
|
||
page_size = clamp_page_size(payload.get("pageSize"), default=DEFAULT_PAGE_SIZE)
|
||
pipeline = ensure_safe_aggregate_pipeline(require_json_array(payload.get("pipeline"), "pipeline", default=[]))
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection = collection_handle(client, database_name, collection_name)
|
||
skip = max(page - 1, 0) * page_size
|
||
|
||
# 先跑计数管道，再跑当前页数据。
|
||
total_result = list(collection.aggregate([*pipeline, {"$count": "total"}]))
|
||
total = int((total_result[0] or {}).get("total") or 0) if total_result else 0
|
||
documents = list(collection.aggregate([*pipeline, {"$skip": skip}, {"$limit": page_size}]))
|
||
rows = [
|
||
{
|
||
"key": f"agg-{page}-{skip + index}",
|
||
"idText": compact_json_preview(document.get("_id"), limit=72) if "_id" in document else "-",
|
||
"idFilterText": dump_extended_json({"_id": document.get("_id")}, pretty=True)
|
||
if "_id" in document
|
||
else "{}",
|
||
"previewText": compact_json_preview(document),
|
||
"jsonText": dump_extended_json(document, pretty=True),
|
||
"document": to_builtin_json(document),
|
||
}
|
||
for index, document in enumerate(documents, start=1)
|
||
]
|
||
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"page": page,
|
||
"pageSize": page_size,
|
||
"total": total,
|
||
"pageCount": max((total + page_size - 1) // page_size, 1),
|
||
"pipelineText": dump_extended_json(pipeline, pretty=True),
|
||
"items": rows,
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def distinct_mongo_values_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = require_name(payload.get("collection"), "collection")
|
||
field_name = require_name(payload.get("field"), "field")
|
||
filter_spec = require_json_object(payload.get("filter"), "filter", default={})
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
values = collection_handle(client, database_name, collection_name).distinct(field_name, filter_spec)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"field": field_name,
|
||
"filterText": dump_extended_json(filter_spec, pretty=True),
|
||
"count": len(values),
|
||
"values": to_builtin_json(values),
|
||
"valuesText": dump_extended_json(values, pretty=True),
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def create_mongo_index_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
index_keys = normalize_index_keys(payload.get("keys"))
|
||
index_options = require_json_object(payload.get("options"), "options", default={})
|
||
|
||
# name 允许前端留空，驱动会自动生成。
|
||
if not str(index_options.get("name") or "").strip():
|
||
index_options.pop("name", None)
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
index_name = collection_handle(client, database_name, collection_name).create_index(index_keys, **index_options)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"indexName": str(index_name),
|
||
"message": "createIndex completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def drop_mongo_index_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
index_name = require_name(payload.get("indexName"), "indexName")
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection_handle(client, database_name, collection_name).drop_index(index_name)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"indexName": index_name,
|
||
"message": "dropIndex completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def drop_mongo_secondary_indexes_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection = collection_handle(client, database_name, collection_name)
|
||
index_names = [
|
||
str(item.get("name") or "").strip()
|
||
for item in collection.list_indexes()
|
||
if str(item.get("name") or "").strip() and str(item.get("name") or "").strip() != "_id_"
|
||
]
|
||
|
||
# Mongo 的 drop_indexes 会保留 _id_ 索引，这里先把删除名单回给前端方便确认结果。
|
||
if not index_names:
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"droppedCount": 0,
|
||
"droppedNames": [],
|
||
"message": "no secondary indexes",
|
||
}
|
||
|
||
collection.drop_indexes()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"droppedCount": len(index_names),
|
||
"droppedNames": index_names,
|
||
"message": "dropIndexes completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def create_mongo_collection_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
options = require_json_object(payload.get("options"), "options", default={})
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
client[database_name].create_collection(collection_name, **options)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"message": "createCollection completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def rename_mongo_collection_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
source_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
target_name = ensure_mutable_collection_name(payload.get("targetCollection"))
|
||
drop_target = bool(payload.get("dropTarget"))
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection_handle(client, database_name, source_name).rename(target_name, dropTarget=drop_target)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": source_name,
|
||
"targetCollection": target_name,
|
||
"message": "renameCollection completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def drop_mongo_collection_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
collection_name = ensure_mutable_collection_name(payload.get("collection"))
|
||
confirm_name = require_name(payload.get("confirmName"), "confirmName")
|
||
|
||
# 删除集合前要求前端把目标名再传一遍，减少误触。
|
||
if confirm_name != collection_name:
|
||
raise ValueError("confirmName does not match collection")
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
collection_handle(client, database_name, collection_name).drop()
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"collection": collection_name,
|
||
"message": "dropCollection completed",
|
||
}
|
||
finally:
|
||
client.close()
|
||
|
||
|
||
def drop_mongo_database_payload(payload: dict[str, Any]) -> dict[str, Any]:
|
||
database_name = require_name(payload.get("database"), "database")
|
||
confirm_name = require_name(payload.get("confirmName"), "confirmName")
|
||
|
||
# 删除数据库同样要求二次校验目标名。
|
||
if confirm_name != database_name:
|
||
raise ValueError("confirmName does not match database")
|
||
|
||
client = get_mongo_client()
|
||
|
||
try:
|
||
client.drop_database(database_name)
|
||
return {
|
||
"ok": True,
|
||
"updatedAt": utc_now(),
|
||
"database": database_name,
|
||
"message": "dropDatabase completed",
|
||
}
|
||
finally:
|
||
client.close()
|