/ /

Tipos personalizados

Página inicial do Docs

/ /

PyMongo

Tipos personalizados

Página inicial do Docs

Bibliotecas do cliente

Python

PyMongo

Tipos personalizados

Codificar dados com codecs de tipo

Visão geral

Este guia explica como usar o PyMongo para codificar e decodificar tipos personalizados.

Codecs de tipo padrão

O PyMongo inclui vários codecs de tipo integrado que você pode usar opcionalmente. Esses codecs de tipo integrados manipulam tipos de dados comuns automaticamente. Por exemplo, o driver fornece classes DecimalEncoder e DecimalDecoder que permitem a codificação e a decodificação do tipo decimal.Decimal do Python de e para valores BSON decimal128.

A classe DecimalEncoder converte valores Python decimal.Decimal em valores BSON Decimal128. A classe DecimalDecoder converte valores BSON Decimal128 em valores Python decimal.Decimal.

O seguinte código utiliza a classe DecimalEncoder para codificar o decimal 1.0:

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

O seguinte código utiliza a classe DecimalDecoder para decodificar dados BSON:

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

Codificar um tipo personalizado

Talvez seja necessário definir um tipo personalizado se quiser armazenar um tipo de dados que o driver não pode serializar nativamente. Por exemplo, tentar salvar uma instância do Enum com PyMongo resulta em uma exceção InvalidDocument , como mostrado no seguinte exemplo de código . Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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'>

As seções seguintes mostram como definir um tipo personalizado para este tipo do Enum .

Definir uma classe de codec de tipo

Para codificar um tipo personalizado, você deve primeiro definir um codec de tipo. Um codec de tipo descreve como uma instância de um tipo personalizado é convertida em e de um tipo que o módulo bson já pode codificar.

Quando você define um codec de tipo, sua classe deve herdar de uma das classes de base no módulo codec_options . A tabela a seguir descreve essas classes básicas e quando e como implementá-las:

Classe base	Quando usar	Membros a serem implementados
`codec_options.TypeEncoder`	Herde desta classe para definir um codec que codifica um tipo Python personalizado para um tipo BSON conhecido.	`python_type` atributo: o tipo de Python personalizado codificado por esse tipo de codec `transform_python()` método: Função que transforma um valor de tipo personalizado em um tipo que o BSON pode codificar
`codec_options.TypeDecoder`	Herde desta classe para definir um codec que decodifica um tipo BSON especificado em um tipo Python personalizado.	`bson_type` atributo: o tipo BSON decodificado por esse codec de tipo `transform_bson()` método: Função que transforma um valor de tipo BSON padrão no tipo personalizado
`codec_options.TypeCodec`	Herde desta classe para definir um codec que possa codificar e decodificar um tipo personalizado.	`python_type` atributo: o tipo de Python personalizado codificado por esse tipo de codec `bson_type` atributo: o tipo BSON decodificado por esse codec de tipo `transform_bson()` método: Função que transforma um valor de tipo BSON padrão no tipo personalizado `transform_python()` método: Função que transforma um valor de tipo personalizado em um tipo que o BSON pode codificar

Como o tipo personalizado do exemplo EnumCodec pode ser convertido de e para uma instância do str , você deve definir como codificar e decodificar esse tipo. Portanto, a classe de codec do tipo Enum deve herdar da classe base 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

Adicionar codec ao registro de tipo

Depois de definir um codec de tipo personalizado, você deve adicioná-lo ao registro de tipos do PyMongo, a lista de tipos que o driver pode codificar e decodificar. Para fazer isso, crie uma instância da classe TypeRegistry , passando uma instância da sua classe de codec do tipo dentro de uma lista. Se você criar vários codecs personalizados, poderá passá-los todos para o construtor TypeRegistry .

Os seguintes exemplos de código adicionam uma instância do codec do tipo EnumCodec ao registro do tipo:

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

Observação

Uma vez instanciados, os registros são imutáveis e a única maneira de adicionar codecs a um registro é criar um novo.

Obter uma referência de collection

Por fim, defina uma instância codec_options.CodecOptions , passando seu objeto TypeRegistry como um argumento de palavra-chave. Passe seu objeto CodecOptions para o método get_collection() para obter uma coleção que possa usar seu tipo personalizado:

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

Você pode então codificar e decodificar instâncias da classe Status . Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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'>}

Para ver como o MongoDB armazena uma instância do tipo personalizado, crie um novo objeto de coleção sem as opções de codec personalizado e use-o para recuperar o documento que contém o tipo personalizado. O exemplo seguinte mostra que o PyMongo armazena uma instância da classe Status como um valor str. Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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'}

Codificar um subtipo

Talvez você também precise codificar um ou mais tipos que herdam do seu tipo personalizado. Considere o seguinte subtipo da classe de enumeração Status , que contém um método para verificar se o status representa um estado ativo:

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

Se você tentar salvar uma instância da classe ExtendedStatus sem primeiro registrar um codec de tipo para ela, o PyMongo chamará um erro. Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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'>

Para codificar uma instância da classe ExtendedStatus , você deve definir um codec de tipo para a classe. Esse codec de tipo deve herdar do codec da classe pai, EnumCodec, conforme mostrado no exemplo a seguir:

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

Em seguida, você pode adicionar o codec de tipo da subclasse ao registro de tipo e codificar instâncias do tipo personalizado. Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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'>}

Observação

O método transform_bson() da classe EnumCodec resulta em esses valores sendo decodificados como Status, não ExtendedStatus.

Definir um codificador de contingência

Você também pode registrar um codificador de contingência, um chamável para codificar tipos não reconhecidos pelo BSON e para os quais nenhum codec de tipo foi registrado. O codificador fallback aceita um valor não codificado como parâmetro e retorna um valor codificado por BSON.

O seguinte codificador de contingência codifica o tipo Enum do Python para str:

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

Depois de declarar um codificador de contingência, execute as seguintes etapas:

Construa uma nova instância da classe TypeRegistry . Use o argumento de palavra-chave fallback_encoder para passar o codificador de contingência.
Construa uma nova instância da classe CodecOptions . Use o argumento da palavra-chave type_registry para passar na instância do TypeRegistry .
Chame o método get_collection() . Use o argumento da palavra-chave codec_options para passar na instância do CodecOptions .

O seguinte exemplo de código mostra este processo:

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

Você pode então usar essa referência a uma coleção para armazenar instâncias da classe Status . Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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'}

Observação

Os codificadores de fallback são invocados após as tentativas de codificar o valor fornecido com codificadores BSON padrão e qualquer codificador de tipo configurado falhar. Portanto, em um registro de tipo configurado com um codificador de tipo e um codificador de fallback que ambos têm como alvo o mesmo tipo personalizado, o comportamento especificado no codificador de tipo tem precedência.

Codificar tipos desconhecidos

Como os codificadores de contingência não precisam declarar os tipos que codificam de antetualmente, você pode usá-los nos casos em que um TypeEncoder não funcione. Por exemplo, você pode usar um codificador de contingência para salvar objetos arbitrários no MongoDB. Considere os seguintes tipos personalizados arbitrários:

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,)

Você pode definir um codificador de contingência que lida com instâncias de enumeração convertendo-as em seus valores de string ou seleciona outros objetos para armazenamento. O exemplo a seguir mostra como lidar com diferentes tipos de objetos personalizados:

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

Em seguida, você pode usar o codificador de contingência em um registro de tipo para codificar e decodificar seus tipos personalizados. Selecione a aba Synchronous ou Asynchronous para ver o código correspondente:

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

Limitações

Os codecs do tipo PyMongo e codificadores de contingência têm as seguintes limitações:

Você não pode personalizar o comportamento de codificação dos tipos de Python que o PyMongo já entende, como int e str. Se você tentar instanciar um registro de tipo com um ou mais codecs que agem em um tipo integrado, o PyMongo gerará um TypeError. Esta limitação também se aplica a todos os subtipos dos tipos padrão.
Você não pode encadear codificadores de tipo. Um valor de tipo personalizado, uma vez transformado pelo método transform_python() de um codec, deve resultar em um tipo que seja codificado por BSON por padrão, ou possa ser transformado pelo codificador de fallback em algo codificado por BSON. Não pode ser transformado uma segunda vez por um codec de tipo diferente.
O método Database.command() não aplica decodificadores de tipo personalizado ao decodificar o documento de resposta do comando.
A classe gridfs não aplica codificação ou decodificação de tipo personalizado a nenhum documento que recebe ou retorna.

Documentação da API

Para obter mais informações sobre codificação e decodificação de tipos personalizados, consulte a seguinte documentação da API:

Voltar

Serialização

Datas e horários