Python查询MongoDB返回含$前缀字段，如何转为常规格式？

针对查询返回的$oid、$date、$binary等带$的BSON字段，以下是几种高效的解决方案：

方案1：使用MongoDB聚合管道在查询时直接转换（推荐） #

通过聚合管道的$project阶段，直接在数据库端完成类型转换，返回的结果无需客户端额外处理：

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017/")
db = client["your_database"]
collection = db["your_collection"]

# 定义聚合管道
pipeline = [
    {
        "$project": {
            "_id": {"$toString": "$_id"},  # 将ObjectId转为字符串
            "createdAt": {"$dateToString": {"format": "%Y-%m-%dT%H:%M:%S.%LZ", "date": "$createdAt"}},  # 转ISO格式日期字符串
            "entityId": {"$uuidToString": "$entityId"},  # 针对UUID类型二进制数据直接转字符串
            "other_field": 1  # 保留其他需要的字段，1表示保留
        }
    }
]

# 执行聚合查询
processed_results = list(collection.aggregate(pipeline))

优势：数据库端处理转换，减少客户端计算压力，批量查询时性能最优，返回结果为普通字典，无特殊$键。

方案2：使用PyMongo自定义BSON解码器 #

如果不想修改查询语句，可通过自定义解码器在客户端自动转换BSON类型：

from pymongo import MongoClient, CodecOptions
from bson.objectid import ObjectId
from bson.binary import Binary
import uuid
from datetime import datetime

def custom_bson_decoder(dct):
    # 转换ObjectId
    if "$oid" in dct:
        return str(ObjectId(dct["$oid"]))
    # 转换日期为ISO字符串
    elif "$date" in dct:
        return datetime.fromtimestamp(dct["$date"] / 1000).isoformat()
    # 转换UUID类型二进制数据（subtype 0x04）
    elif "$binary" in dct and dct["$type"] == "04":
        return str(uuid.UUID(bytes=dct["$binary"]))
    # 其他类型保留默认处理
    else:
        return dct

client = MongoClient("mongodb://localhost:27017/")
db = client["your_database"]
# 设置解码器选项
codec_options = CodecOptions(document_class=dict, tz_aware=True)
collection = db["your_collection"].with_options(codec_options=codec_options)

# 查询后应用自定义解码器
raw_results = list(collection.find({}))
processed_results = [custom_bson_decoder(doc) for doc in raw_results]

优势：无需修改查询逻辑，解码器可灵活扩展处理更多BSON类型。

方案3：查询后批量处理特定字段 #

针对小批量数据，可直接遍历结果处理目标字段，逻辑清晰直观：

from pymongo import MongoClient
from bson.binary import Binary
import uuid
from datetime import datetime

client = MongoClient("mongodb://localhost:27017/")
db = client["your_database"]
collection = db["your_collection"]

raw_results = list(collection.find({}))
processed_results = []

for doc in raw_results:
    processed_doc = doc.copy()
    # 处理ObjectId
    processed_doc["_id"] = str(processed_doc["_id"])
    # 处理日期
    if isinstance(processed_doc["createdAt"], datetime):
        processed_doc["createdAt"] = processed_doc["createdAt"].isoformat()
    # 处理UUID类型二进制字段
    if isinstance(processed_doc["entityId"], Binary) and processed_doc["entityId"].subtype == 4:
        processed_doc["entityId"] = str(uuid.UUID(bytes=processed_doc["entityId"]))
    processed_results.append(processed_doc)

注意事项：

• 若entityId的二进制 subtype 不是0x04（UUID标准类型），需调整转换逻辑适配对应 subtype。
• 日期格式可通过$dateToString的format参数（方案1）或Python的strftime（方案3）自定义。

内容的提问来源于stack exchange，提问作者thankYou