/ /

/ /

使用类型编解码器对数据进行编码

Overview

本指南介绍了如何使用 PyMongo 对自定义类型进行编码和解码。

默认类型编解码器

PyMongo包含多个内置类型编解码器，您可以选择使用。这些内置类型编解码器会自动处理常见数据类型。示例，驱动程序提供了 DecimalEncoder 和 DecimalDecoder 类，启用Python 的 decimal.Decimal 类型与BSON Decimal128 值进行编码和解码。

DecimalEncoder 类将Python decimal.Decimal 值转换为BSON Decimal128 值。DecimalDecoder 类将BSON Decimal128 值转换为Python decimal.Decimal 值。

以下代码使用 DecimalEncoder 类对十进制 1.0 进行编码：

opts = CodecOptions(type_registry=TypeRegistry([DecimalEncoder()]))
bson.encode({"d": decimal.Decimal('1.0')}, codec_options=opts)

以下代码使用 DecimalDecoder 类来解码BSON数据：

opts = CodecOptions(type_registry=TypeRegistry([DecimalDecoder()]))
bson.decode(data, codec_options=opts)

对自定义类型进行编码

如果要存储驱动程序无法本机序列化的数据类型，则可能需要定义自定义类型。例如，尝试使用 PyMongo 保存 Enum 的实例会导致 InvalidDocument 异常，如以下代码示例所示。选择 Synchronous 或 Asynchronous标签页以查看相应的代码：

from enum import Enum
class Status(Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
status = Status.ACTIVE
db["coll"].insert_one({"status": status})

Traceback (most recent call last):
...
bson.errors.InvalidDocument: Invalid document {'status': <Status.ACTIVE:
'active'>, '_id': ObjectId('68bb6144862a5cfb94a9fd48')} | cannot encode
object: <Status.ACTIVE: 'active'>, of type: <enum 'Status'>

from enum import Enum
class Status(Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
status = Status.ACTIVE
await db["coll"].insert_one({"status": status})

Traceback (most recent call last):
...
bson.errors.InvalidDocument: Invalid document {'status': <Status.ACTIVE:
'active'>, '_id': ObjectId('68bb6144862a5cfb94a9fd48')} | cannot encode
object: <Status.ACTIVE: 'active'>, of type: <enum 'Status'>

以下部分介绍如何为此Enum类型定义自定义类型。

定义类型编解码器类

要对自定义类型进行编码，必须首先定义类型编解码器。 类型编解码器描述了如何将自定义类型的实例与bson模块已可编码的类型相互转换。

定义类型编解码器时，您的类必须从codec_options模块中的基类之一继承。下表描述了这些基类，以及何时以及如何实现它们：

基类	何时使用	要实施的成员
`codec_options.TypeEncoder`	从此类继承以定义编解码器，将自定义 Python 类型编码为已知的 BSON 类型。	`python_type` 属性：由此类型编解码器编码的自定义 Python 类型 `transform_python()` 方法：将自定义类型值转换为 BSON 可以编码的类型的函数
`codec_options.TypeDecoder`	从此类继承以定义编解码器，将指定的 BSON 类型解码为自定义 Python 类型。	`bson_type` 属性：由此类型编解码器解码的 BSON 类型 `transform_bson()` 方法：将标准 BSON 类型值转换为自定义类型的函数
`codec_options.TypeCodec`	从此类继承以定义可对自定义类型进行编码和解码的编解码器。	`python_type` 属性：由此类型编解码器编码的自定义 Python 类型 `bson_type` 属性：由此类型编解码器解码的 BSON 类型 `transform_bson()` 方法：将标准 BSON 类型值转换为自定义类型的函数 `transform_python()` 方法：将自定义类型值转换为 BSON 可以编码的类型的函数

由于示例的EnumCodec自定义类型可以与str实例相互转换，因此您必须定义如何对此类型进行编码和解码。因此， Enum类型编解码器类必须从TypeCodec基类继承：

from bson.codec_options import TypeCodec
class EnumCodec(TypeCodec):
    python_type = Status
    bson_type = str
    def transform_python(self, value):
        return value.value
    def transform_bson(self, value):
        try:
            return Status(value)
        except ValueError:
            return value

将编解码器添加到类型注册表

定义自定义类型编解码器后，必须将其添加到 PyMongo 的类型注册表，即驱动程序可以编码和解码的类型列表。为此，创建TypeRegistry类的实例，传入列表中类型编解码器类的实例。如果创建多个自定义编解码器，则可以将它们全部传递给TypeRegistry构造函数。

以下代码示例将EnumCodec类型编解码器的实例添加到类型注册表中：

from bson.codec_options import TypeRegistry
enum_codec = EnumCodec()
type_registry = TypeRegistry([enum_codec])

注意

一旦实例化，注册表就不可变，将编解码器添加到注册表的唯一方法是创建一个新的编解码器。

获取集合引用

最后，定义一个codec_options.CodecOptions实例，将TypeRegistry对象作为关键字参数传递。将CodecOptions对象传递给get_collection()方法，以获得可以使用自定义类型的集合：

from bson.codec_options import CodecOptions
codec_options = CodecOptions(type_registry=type_registry)
collection = database.get_collection("test", codec_options=codec_options)

然后，您可以对 Status 类的实例进行编码和解码。选择Synchronous或Asynchronous标签页以查看相应的代码：

import pprint
collection.insert_one({"status": Status.ACTIVE})
my_doc = collection.find_one({"status": "active"})
pprint.pprint(my_doc)

{'_id': ObjectId('...'), 'status': <Status.ACTIVE: 'active'>}

import pprint
await collection.insert_one({"status": Status.ACTIVE})
my_doc = await collection.find_one({"status": "active"})
pprint.pprint(my_doc)

{'_id': ObjectId('...'), 'status': <Status.ACTIVE: 'active'>}

要查看MongoDB如何存储自定义类型的实例，请创建一个不带自定义编解码器选项的新集合对象，然后使用它来检索包含自定义类型的文档。以下示例显示PyMongo将 Status 类的实例存储为 str 值。选择Synchronous或Asynchronous标签页以查看相应的代码：

import pprint
new_collection = database.get_collection("test")
pprint.pprint(new_collection.find_one())

{'_id': ObjectId('...'), 'status': 'active'}

import pprint
new_collection = database.get_collection("test")
pprint.pprint(await new_collection.find_one())

{'_id': ObjectId('...'), 'status': 'active'}

对子类型进行编码

您可能还需要对从自定义类型继承的一个或多个类型进行编码。考虑 Status枚举类的以下子类型，其中包含检查状态是否表示活动状态的方法：

class ExtendedStatus(Enum):
    ACTIVE = "active"
    INACTIVE = "inactive"
    PENDING = "pending"
    def is_active(self):
        return self == ExtendedStatus.ACTIVE

如果您在没有先为其注册类型编解码器的情况下尝试保存 ExtendedStatus 类的实例， PyMongo会引发错误。选择Synchronous或Asynchronous标签页以查看相应的代码：

collection.insert_one({"status": ExtendedStatus.ACTIVE})

Traceback (most recent call last):
...
bson.errors.InvalidDocument: Invalid document {'status': <ExtendedStatus.ACTIVE: 'active'>, '_id': ObjectId('...')} | cannot encode object: <ExtendedStatus.ACTIVE: 'active'>, of type: <enum 'ExtendedStatus'>

await collection.insert_one({"status": ExtendedStatus.ACTIVE})

Traceback (most recent call last):
...
bson.errors.InvalidDocument: Invalid document {'status': <ExtendedStatus.ACTIVE: 'active'>, '_id': ObjectId('...')} | cannot encode object: <ExtendedStatus.ACTIVE: 'active'>, of type: <enum 'ExtendedStatus'>

要对ExtendedStatus类的实例进行编码，您必须为该类定义类型编解码器。此类型编解码器必须从父类的编解码器EnumCodec继承，如以下示例所示：

class ExtendedStatusCodec(EnumCodec):
    @property
    def python_type(self):
        # The Python type encoded by this type codec
        return ExtendedStatus

然后，您可以将子类的类型编解码器添加到类型注册表中，并对自定义类型的实例进行编码。选择 Synchronous 或 Asynchronous标签页，查看相应的代码：

import pprint
from bson.codec_options import CodecOptions
extended_status_codec = ExtendedStatusCodec()
type_registry = TypeRegistry([enum_codec, extended_status_codec])
codec_options = CodecOptions(type_registry=type_registry)
collection = database.get_collection("test", codec_options=codec_options)
collection.insert_one({"status": ExtendedStatus.ACTIVE})
my_doc = collection.find_one()
pprint.pprint(my_doc)

{'_id': ObjectId('...'), 'status': <Status.ACTIVE: 'active'>}

import pprint
from bson.codec_options import CodecOptions
extended_status_codec = ExtendedStatusCodec()
type_registry = TypeRegistry([enum_codec, extended_status_codec])
codec_options = CodecOptions(type_registry=type_registry)
collection = database.get_collection("test", codec_options=codec_options)
await collection.insert_one({"status": ExtendedStatus.ACTIVE})
my_doc = await collection.find_one()
pprint.pprint(my_doc)

{'_id': ObjectId('...'), 'status': <Status.ACTIVE: 'active'>}

注意

EnumCodec类的transform_bson()方法会导致这些值解码为Status ，而不是ExtendedStatus 。

定义回退编码器

您还可以注册一个回退编码器，这是一个可调用函数，用于对 BSON 无法识别且尚未为其注册类型编解码器的类型进行编码。回退编码器接受不可编码的值作为参数，并返回 BSON 可编码的值。

以下回退编码器将 Python 的Enum类型编码为str ：

from enum import Enum
def fallback_encoder(value):
    if isinstance(value, Enum):
        return value.value
    return value

声明回退编码器后，执行以下步骤：

构造TypeRegistry类的新实例。使用fallback_encoder关键字参数传入回退编码器。
构造CodecOptions类的新实例。使用type_registry关键字参数传入TypeRegistry实例。
调用get_collection()方法。使用codec_options关键字参数传入CodecOptions实例。

以下代码示例展示了此过程：

type_registry = TypeRegistry(fallback_encoder=fallback_encoder)
codec_options = CodecOptions(type_registry=type_registry)
collection = db.get_collection("test", codec_options=codec_options)

然后，您可以使用这个对集合的引用来存储Status 类的实例。选择Synchronous或Asynchronous标签页以查看相应的代码：

import pprint
collection.insert_one({"status": Status.ACTIVE})
my_doc = collection.find_one()
pprint.pprint(my_doc)

{'_id': ObjectId('...'), 'status': 'active'}

import pprint
await collection.insert_one({"status": Status.ACTIVE})
my_doc = await collection.find_one()
pprint.pprint(my_doc)

{'_id': ObjectId('...'), 'status': 'active'}

注意

在尝试使用标准 BSON 编码器和任何配置类型编码器对给定值进行编码失败后，会调用回退编码器。因此，在配置了针对同一自定义类型的类型编码器和回退编码器的类型注册表中，类型编码器中指定的行为优先。

对未知类型进行编码

由于回退编码器无需事先声明其编码的类型，因此可在TypeEncoder不起作用的情况下使用回退编码器。例如，您可以使用回退编码器将任意对象保存到 MongoDB。考虑以下任意自定义类型：

class MyStringType(object):
    def __init__(self, value):
        self.__value = value
    def __repr__(self):
        return "MyStringType('%s')" % (self.__value,)
class MyNumberType(object):
    def __init__(self, value):
        self.__value = value
    def __repr__(self):
        return "MyNumberType(%s)" % (self.__value,)

您可以定义一个回退编码器，通过将枚举实例转换为字符串值来处理枚举实例，或对其他对象进行序列化以进行存储。以下示例展示了如何处理不同类型的自定义对象：

import pickle
from enum import Enum
def fallback_pickle_encoder(value):
    if isinstance(value, Enum):
        return value.value
    return pickle.dumps(value).decode('latin-1')
class PickledStringDecoder(TypeDecoder):
    bson_type = str
    def transform_bson(self, value):
        try:
            # Try to unpickle the string value
            return pickle.loads(value.encode('latin-1'))
        except:
            # If unpickling fails, return the original string
            return value

然后，您可以使用类型注册表中的回退编码器对自定义类型进行编码和解码。选择 Synchronous 或 Asynchronous标签页，查看相应的代码：

from bson.codec_options import CodecOptions,TypeRegistry
codec_options = CodecOptions(
   type_registry=TypeRegistry(
      fallback_encoder=fallback_pickle_encoder
)
)
collection = db.get_collection("test", codec_options=codec_options)
collection.insert_one(
   {"_id": 1, "str": MyStringType("hello world"), "num": MyNumberType(2)}
)
my_doc = collection.find_one()
print(isinstance(my_doc["str"], MyStringType))
print(isinstance(my_doc["num"], MyNumberType))

True
True

from bson.codec_options import CodecOptions,TypeRegistry
codec_options = CodecOptions(
   type_registry=TypeRegistry(
      fallback_encoder=fallback_pickle_encoder
)
)
collection = db.get_collection("test", codec_options=codec_options)
await collection.insert_one(
   {"_id": 1, "str": MyStringType("hello world"), "num": MyNumberType(2)}
)
my_doc = await collection.find_one()
print(isinstance(my_doc["str"], MyStringType))
print(isinstance(my_doc["num"], MyNumberType))

True
True

限制

PyMongo 类型编解码器和回退编码器具有以下限制：

您无法自定义 PyMongo 已理解的 Python 类型的编码行为，例如int和str 。如果尝试使用一个或多个作用于内置类型的编解码器实例化类型注册表，PyMongo 会引发TypeError 。此限制也适用于标准类型的所有子类型。
您不能链接类型编码器。自定义类型值一旦被编解码器的transform_python()方法转换，就必须默认生成 BSON 可编码的类型，或者可以由回退编码器转换为可 BSON 编码的类型。它不能由不同类型的编解码器再次转换。
Database.command()方法在解码命令响应文档时不应用自定义类型解码器。
gridfs类不会对其接收或返回的任何文档应用自定义类型编码或解码。

API 文档

有关对自定义类型进行编码和解码的更多信息，请参阅以下 API 文档：

后退

序列化

来年

日期和时间