Skip to content

# odmantic.bson

This module provides helpers to build Pydantic Models containing BSON objects.

## Pydantic model helpers

Bases: `BaseModel`

Equivalent of `pydantic.BaseModel` supporting BSON types serialization.

If you want to apply other custom JSON encoders, you'll need to use BSON_TYPES_ENCODERS directly.

Source code in `odmantic/bson.py`

    418
    419
    420
    421
    422
    423
    424
    425

    class BaseBSONModel(BaseModel):
        """Equivalent of `pydantic.BaseModel` supporting BSON types serialization.

        If you want to apply other custom JSON encoders, you'll need to use
        [BSON_TYPES_ENCODERS][odmantic.bson.BSON_TYPES_ENCODERS] directly.
        """

        model_config = {"json_encoders": BSON_TYPES_ENCODERS}

Encoders required to encode BSON fields (can be used in the Pydantic Model's `Config.json_encoders` parameter). See pydantic: JSON Encoders for more details.

## Custom BSON serializer annotation

Adds a BSON serializer to use on a field when it will be saved to the database

Source code in `odmantic/bson.py`

    23
    24
    25
    26
    27

    @dataclass(frozen=True)
    class WithBsonSerializer:
        """Adds a BSON serializer to use on a field when it will be saved to the database"""

        bson_serializer: Callable[[Any], Any]

## Pydantic type helpers

Those helpers inherit directly from their respective `bson` types. They add the field validation logic required by Pydantic to work with them. On top of this, the appropriate JSON schemas are generated for them.

Bases: `ObjectId`

Source code in `odmantic/bson.py`

    40
    41
    42
    43
    44
    45
    46
    47
    48
    49
    50
    51
    52
    53
    54
    55
    56
    57
    58
    59
    60
    61
    62
    63
    64
    65
    66
    67
    68
    69
    70
    71
    72
    73
    74
    75
    76
    77
    78
    79
    80
    81
    82
    83
    84
    85
    86
    87
    88
    89

    class ObjectId(bson.ObjectId):
        @classmethod
        def __get_pydantic_core_schema__(
            cls,
            _source_type: Any,
            _handler: Callable[[Any], core_schema.CoreSchema],
        ) -> core_schema.CoreSchema:
            def validate_from_string_or_bytes(value: Union[str, bytes]) -> bson.ObjectId:
                try:
                    return bson.ObjectId(value)
                except bson.errors.InvalidId:
                    raise ValueError("Invalid ObjectId")

            from_string_or_bytes_schema = core_schema.chain_schema(
                [
                    core_schema.union_schema(
                        [
                            core_schema.str_schema(),
                            core_schema.bytes_schema(),
                        ]
                    ),
                    core_schema.no_info_plain_validator_function(
                        validate_from_string_or_bytes
                    ),
                ]
            )

            return core_schema.json_or_python_schema(
                json_schema=from_string_or_bytes_schema,
                python_schema=core_schema.union_schema(
                    [
                        core_schema.is_instance_schema(bson.ObjectId),
                        from_string_or_bytes_schema,
                    ],
                ),
                serialization=core_schema.plain_serializer_function_ser_schema(
                    str, when_used="json"
                ),
            )

        @classmethod
        def __get_pydantic_json_schema__(
            cls, _schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
        ) -> JsonSchemaValue:
            json_schema = handler(core_schema.str_schema())
            json_schema.update(
                examples=["5f85f36d6dfecacc68428a46", "ffffffffffffffffffffffff"],
                example="5f85f36d6dfecacc68428a46",
            )
            return json_schema

Bases: `Int64`

Source code in `odmantic/bson.py`

     92
     93
     94
     95
     96
     97
     98
     99
    100
    101
    102
    103
    104
    105
    106
    107
    108
    109
    110
    111
    112
    113
    114
    115
    116
    117
    118
    119
    120
    121
    122
    123
    124

    class Int64(bson.Int64):
        @classmethod
        def __get_pydantic_core_schema__(
            cls,
            _source_type: Any,
            _handler: Callable[[Any], core_schema.CoreSchema],
        ) -> core_schema.CoreSchema:
            def validate_from_int(value: int) -> bson.int64.Int64:
                return bson.int64.Int64(value)

            from_int_schema = core_schema.chain_schema(
                [
                    core_schema.int_schema(),
                    core_schema.no_info_plain_validator_function(validate_from_int),
                ]
            )

            return core_schema.json_or_python_schema(
                json_schema=from_int_schema,
                python_schema=core_schema.union_schema(
                    [
                        core_schema.is_instance_schema(bson.int64.Int64),
                        from_int_schema,
                    ]
                ),
            )

        @classmethod
        def __get_pydantic_json_schema__(
            cls, _core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
        ) -> JsonSchemaValue:
            # Use the same schema that would be used for `int`
            return handler(core_schema.int_schema())

Bases: `Decimal128`

Source code in `odmantic/bson.py`

    130
    131
    132
    133
    134
    135
    136
    137
    138
    139
    140
    141
    142
    143
    144
    145
    146
    147
    148
    149
    150
    151
    152
    153
    154
    155
    156
    157
    158
    159
    160
    161
    162
    163
    164
    165

    class Decimal128(bson.decimal128.Decimal128):
        @classmethod
        def __get_pydantic_core_schema__(
            cls,
            _source_type: Any,
            _handler: Callable[[Any], core_schema.CoreSchema],
        ) -> core_schema.CoreSchema:
            def validate_from_decimal_repr(
                value: Union[decimal.Decimal, float, str, Tuple[int, Sequence[int], int]],
            ) -> bson.decimal128.Decimal128:
                try:
                    return bson.decimal128.Decimal128(value)
                except Exception:
                    raise ValueError("Invalid Decimal128 value")

            from_decimal_repr_schema = core_schema.no_info_plain_validator_function(
                validate_from_decimal_repr
            )
            return core_schema.json_or_python_schema(
                json_schema=from_decimal_repr_schema,
                python_schema=core_schema.union_schema(
                    [
                        core_schema.is_instance_schema(bson.decimal128.Decimal128),
                        from_decimal_repr_schema,
                    ]
                ),
                serialization=core_schema.plain_serializer_function_ser_schema(
                    lambda v: v.to_decimal(), when_used="json"
                ),
            )

        @classmethod
        def __get_pydantic_json_schema__(
            cls, _core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
        ) -> JsonSchemaValue:
            return handler(core_schema.float_schema())

Bases: `Binary`

Source code in `odmantic/bson.py`

    168
    169
    170
    171
    172
    173
    174
    175
    176
    177
    178
    179
    180
    181
    182
    183
    184
    185
    186
    187
    188
    189
    190
    191
    192
    193
    194
    195
    196
    197
    198
    199
    200

    class Binary(bson.binary.Binary):
        @classmethod
        def __get_pydantic_core_schema__(
            cls,
            _source_type: Any,
            _handler: Callable[[Any], core_schema.CoreSchema],
        ) -> core_schema.CoreSchema:
            def validate_from_bytes(
                value: bytes,
            ) -> bson.binary.Binary:
                return bson.binary.Binary(value)

            from_bytes_schema = core_schema.chain_schema(
                [
                    core_schema.bytes_schema(),
                    core_schema.no_info_plain_validator_function(validate_from_bytes),
                ]
            )
            return core_schema.json_or_python_schema(
                json_schema=from_bytes_schema,
                python_schema=core_schema.union_schema(
                    [
                        core_schema.is_instance_schema(bson.binary.Binary),
                        from_bytes_schema,
                    ]
                ),
            )

        @classmethod
        def __get_pydantic_json_schema__(
            cls, _core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
        ) -> JsonSchemaValue:
            return handler(core_schema.bytes_schema())

Bases: `Regex`

Source code in `odmantic/bson.py`

    230
    231
    232
    233
    234
    235
    236
    237
    238
    239
    240
    241
    242
    243
    244
    245
    246
    247
    248
    249
    250
    251
    252
    253
    254
    255
    256
    257
    258
    259
    260
    261
    262
    263
    264
    265
    266
    267
    268
    269
    270
    271
    272
    273
    274
    275
    276

    class Regex(bson.regex.Regex):
        @classmethod
        def __get_pydantic_core_schema__(
            cls,
            _source_type: Any,
            _handler: Callable[[Any], core_schema.CoreSchema],
        ) -> core_schema.CoreSchema:
            from_str_schema = core_schema.chain_schema(
                [
                    core_schema.str_schema(),
                    core_schema.no_info_plain_validator_function(validate_pattern_from_str),
                    core_schema.no_info_plain_validator_function(
                        validate_regex_from_pattern
                    ),
                ]
            )
            from_pattern_schema = core_schema.chain_schema(
                [
                    core_schema.is_instance_schema(Pattern),
                    core_schema.no_info_plain_validator_function(
                        validate_regex_from_pattern
                    ),
                ]
            )
            return core_schema.json_or_python_schema(
                json_schema=from_str_schema,
                python_schema=core_schema.union_schema(
                    [
                        core_schema.is_instance_schema(bson.regex.Regex),
                        from_pattern_schema,
                        from_str_schema,
                    ]
                ),
                serialization=core_schema.plain_serializer_function_ser_schema(
                    lambda v: v.pattern, when_used="json"
                ),
            )

        @classmethod
        def __get_pydantic_json_schema__(
            cls, _core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler
        ) -> JsonSchemaValue:
            schema = handler(core_schema.str_schema())
            schema.update(
                examples=[r"^Foo"], example=r"^Foo", type="string", format="binary"
            )
            return schema