유형 코덱으로 데이터 인코딩

기본 유형 코덱

PyMongo 에는 선택적으로 사용할 수 있는 몇 가지 내장 유형 코덱이 포함되어 있습니다. 이러한 내장 유형 코덱은 일반적인 데이터 유형을 자동으로 처리하다 . 예시 들어 운전자 Python의 decimal.Decimal 유형을 BSON Decimal128 값으로 인코딩 및 디코딩 활성화 있는 DecimalEncoder 및 DecimalDecoder 클래스를 제공합니다.

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)

유형 코덱 클래스 정의

사용자 지정 유형을 인코딩하려면 먼저 유형 코덱을 정의해야 합니다. 유형 코덱은 사용자 지정 유형의 인스턴스가 bson 모듈이 이미 인코딩할 수 있는 유형으로 변환되거나 그 반대로 변환되는 방법을 설명합니다.

유형 코덱을 정의할 때 클래스는 codec_options 모듈의 기본 클래스 중 하나에서 상속되어야 합니다. 다음 표에서는 이러한 기본 클래스와 이를 구현하는 시기와 방법에 대해 설명합니다.

예시의 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'}

알 수 없는 유형 인코딩

폴백 인코더는 인코딩하는 유형을 미리 선언할 필요가 없으므로 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