Python加载MongoDB中Binary.createFromBase64字段时遇InvalidBSON错误

问题背景 #

MongoDB集合中文档结构如下：

{
  _id: ObjectId('656f001650078dc1d50872ed'),
  created_at: ISODate('2023-12-05T10:48:54.641Z'),
  user_id: Binary.createFromBase64('MjkyMmZkYmUtOTM1Yi0xMWVlLTlkMjEtN2U2NjQwYmEyNGEw', 4),
} 

使用Python代码查询时，投影created_at字段正常，但切换为投影user_id字段时触发InvalidBSON错误。

Python查询代码：

from pymongo import MongoClient
import os

client = MongoClient(os.getenv("MONGO_URL"))
collection = client.get_database(os.getenv("MONGO_DB")).get_collection(os.getenv("MONGO_COLLECTION"))

# 正常运行
select = {'created_at': 1}
result = collection.find_one({}, select)

# 触发错误
select = {'user_id': 1}
result = collection.find_one({}, select)

错误详情 #

File ~/.pyenv/versions/3.11.5/lib/python3.11/site-packages/pymongo/message.py:1619, in _OpMsg.unpack_response(self, cursor_id, codec_options, user_fields, legacy_response)
   1617 # If _OpMsg is in-use, this cannot be a legacy response.
   1618 assert not legacy_response
-> 1619 return bson._decode_all_selective(self.payload_document, codec_options, user_fields)

File ~/.pyenv/versions/3.11.5/lib/python3.11/site-packages/bson/__init__.py:1259, in _decode_all_selective(data, codec_options, fields)
   1236 """Decode BSON data to a single document while using user-provided
   1237 custom decoding logic.
   1238 
   (...)
   1256 .. versionadded:: 3.8
   1257 """
   1258 if not codec_options.type_registry._decoder_map:
-> 1259     return decode_all(data, codec_options)

File ~/.pyenv/versions/3.11.5/lib/python3.11/site-packages/bson/__init__.py:1167, in decode_all(data, codec_options)
   1164 if not isinstance(codec_options, CodecOptions):
   1165     raise _CODEC_OPTIONS_TYPE_ERROR
-> 1167 return _decode_all(data, codec_options)

InvalidBSON: invalid length or type code

环境版本 #

• Python: 3.11.5
• pymongo: 4.6.1
• MongoDB: 5.0.23

解决思路 #

1. 检查Binary子类型与数据匹配性 #

Binary.createFromBase64(..., 4)中的子类型4对应UUID（RFC 4122），要求二进制数据长度必须为16字节。从提供的Base64字符串解码后长度为36字节（非UUID标准长度），属于子类型误用。

• 验证数据长度：在MongoDB Shell中执行以下命令，查看user_id的二进制长度：

db.your_collection.find({}, {user_id: 1}).forEach(doc => print(`user_id长度: ${doc.user_id.length}`))

如果长度不是16，说明存储时错误使用了UUID子类型，应改为子类型0（通用二进制）。

2. 自定义解码规则绕过UUID自动解码 #

如果无法修改现有数据的子类型，可以通过CodecOptions强制pymongo将子类型4的Binary当作普通二进制处理：

from pymongo import MongoClient
from bson.codec_options import CodecOptions
import os

# 创建CodecOptions，禁用UUID自动解码
codec_options = CodecOptions(uuid_representation='standard')
client = MongoClient(os.getenv("MONGO_URL"), codec_options=codec_options)

collection = client.get_database(os.getenv("MONGO_DB")).get_collection(os.getenv("MONGO_COLLECTION"))
select = {'user_id': 1}
result = collection.find_one({}, select)
# 此时result['user_id']为bson.binary.Binary对象，可通过result['user_id'].decode('utf-8')获取字符串

3. 在MongoDB端转换数据后再查询 #

使用聚合管道的$convert操作符，在数据库端将Binary转换为字符串，避免Python端解码错误：

from pymongo import MongoClient
import os

client = MongoClient(os.getenv("MONGO_URL"))
collection = client.get_database(os.getenv("MONGO_DB")).get_collection(os.getenv("MONGO_COLLECTION"))

# 使用聚合投影转换user_id
result = collection.aggregate([
    {
        "$project": {
            "user_id": {
                "$convert": {
                    "input": "$user_id",
                    "to": "string",
                    "onError": "$user_id"  # 转换失败时保留原Binary数据
                }
            }
        }
    }
]).next()

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