0
# BSON Types
1
2
ODMantic provides native MongoDB BSON types for precise data representation and serialization. These types ensure proper handling of MongoDB-specific data formats.
3
4
```python
5
from odmantic import ObjectId, WithBsonSerializer
6
from odmantic.bson import Int64, Long, Decimal128, Binary, Regex
7
from decimal import Decimal
8
```
9
10
## Capabilities
11
12
### ObjectId
13
14
MongoDB's primary key type providing unique document identifiers.
15
16
```python { .api }
17
class ObjectId:
18
"""MongoDB ObjectId type for unique document identification."""
19
20
def __init__(self, oid=None):
21
"""
22
Create ObjectId from hex string, bytes, or generate new one.
23
24
Args:
25
oid: Hex string, bytes, or None for auto-generation
26
"""
27
28
def __str__(self):
29
"""
30
String representation as 24-character hex string.
31
32
Returns:
33
str: Hex string representation
34
"""
35
36
def __repr__(self):
37
"""
38
Python representation.
39
40
Returns:
41
str: ObjectId('hex_string')
42
"""
43
44
@property
45
def generation_time(self):
46
"""
47
Time when ObjectId was generated.
48
49
Returns:
50
datetime: Generation timestamp
51
"""
52
53
@classmethod
54
def from_datetime(cls, dt):
55
"""
56
Create ObjectId from datetime.
57
58
Args:
59
dt: datetime object
60
61
Returns:
62
ObjectId: ObjectId with timestamp from datetime
63
"""
64
65
@classmethod
66
def is_valid(cls, oid):
67
"""
68
Check if string/bytes represents valid ObjectId.
69
70
Args:
71
oid: String or bytes to validate
72
73
Returns:
74
bool: True if valid ObjectId format
75
"""
76
```
77
78
### BSON Serialization Support
79
80
Mixin class for custom BSON serialization behavior.
81
82
```python { .api }
83
class WithBsonSerializer:
84
"""Mixin for custom BSON serialization."""
85
86
def __bson__(self):
87
"""
88
Custom BSON serialization method.
89
90
Returns:
91
Any: BSON-serializable representation
92
"""
93
```
94
95
### Extended BSON Types
96
97
Additional MongoDB BSON types for specialized use cases.
98
99
```python { .api }
100
class Int64:
101
"""64-bit integer type for MongoDB."""
102
103
def __init__(self, value: int):
104
"""
105
Create 64-bit integer.
106
107
Args:
108
value: Integer value
109
"""
110
111
Long = Int64 # Alias for Int64
112
113
class Decimal128:
114
"""128-bit decimal type for high-precision numbers."""
115
116
def __init__(self, value: Union[str, int, float, Decimal]):
117
"""
118
Create 128-bit decimal.
119
120
Args:
121
value: Decimal, string, int, or float value
122
"""
123
124
def to_decimal(self) -> Decimal:
125
"""
126
Convert to Python decimal.Decimal.
127
128
Returns:
129
Decimal: Python decimal representation
130
"""
131
132
class Binary:
133
"""Binary data type for MongoDB."""
134
135
def __init__(self, data: bytes, subtype: int = 0):
136
"""
137
Create binary data.
138
139
Args:
140
data: Bytes data
141
subtype: Binary subtype (0-255)
142
"""
143
144
class Regex:
145
"""Regular expression type for MongoDB."""
146
147
def __init__(self, pattern: str, flags: int = 0):
148
"""
149
Create regex.
150
151
Args:
152
pattern: Regular expression pattern string
153
flags: Regex flags (integer)
154
"""
155
156
@property
157
def pattern(self) -> str:
158
"""
159
Get regex pattern.
160
161
Returns:
162
str: Regular expression pattern
163
"""
164
165
@property
166
def flags(self) -> int:
167
"""
168
Get regex flags.
169
170
Returns:
171
int: Regular expression flags
172
"""
173
```
174
175
## Usage Examples
176
177
### ObjectId Usage
178
179
```python
180
from odmantic import Model, ObjectId
181
from datetime import datetime
182
183
class User(Model):
184
name: str
185
# Default ObjectId primary key (automatic)
186
# id: ObjectId is added automatically
187
188
class Post(Model):
189
title: str
190
author_id: ObjectId # Reference to User
191
content: str
192
193
# Creating ObjectIds
194
def objectid_examples():
195
# Auto-generated ObjectId
196
user_id = ObjectId()
197
print(f"Generated: {user_id}")
198
199
# From hex string
200
existing_id = ObjectId("507f1f77bcf86cd799439011")
201
print(f"From string: {existing_id}")
202
203
# From datetime (useful for date-based queries)
204
yesterday = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
205
yesterday_id = ObjectId.from_datetime(yesterday)
206
print(f"From datetime: {yesterday_id}")
207
208
# Validation
209
if ObjectId.is_valid("507f1f77bcf86cd799439011"):
210
print("Valid ObjectId format")
211
212
# Generation time
213
print(f"Generated at: {user_id.generation_time}")
214
215
# Using ObjectIds in models
216
async def model_objectid_example(engine):
217
# Create user with auto-generated ID
218
user = User(name="John Doe")
219
await engine.save(user)
220
print(f"User ID: {user.id}")
221
222
# Create post referencing the user
223
post = Post(
224
title="My First Post",
225
author_id=user.id, # Reference using ObjectId
226
content="Hello world!"
227
)
228
await engine.save(post)
229
230
# Query by ObjectId
231
found_user = await engine.find_one(User, User.id == user.id)
232
user_posts = await engine.find(Post, Post.author_id == user.id)
233
```
234
235
### Custom Primary Keys with ObjectId
236
237
```python
238
class CustomModel(Model):
239
# Custom ObjectId field as primary key
240
custom_id: ObjectId = Field(primary_field=True, default_factory=ObjectId)
241
name: str
242
value: int
243
244
def custom_primary_example():
245
# Model uses custom_id as _id in MongoDB
246
model = CustomModel(name="test", value=42)
247
print(f"Custom ID: {model.custom_id}")
248
249
# Can also set explicitly
250
specific_id = ObjectId("507f1f77bcf86cd799439011")
251
model2 = CustomModel(custom_id=specific_id, name="specific", value=100)
252
```
253
254
### Date-based ObjectId Queries
255
256
```python
257
from datetime import datetime, timedelta
258
259
async def date_based_queries(engine):
260
# Find documents created since yesterday
261
yesterday = datetime.utcnow() - timedelta(days=1)
262
yesterday_id = ObjectId.from_datetime(yesterday)
263
264
recent_posts = await engine.find(Post, Post.id >= yesterday_id)
265
266
# Find documents from specific date range
267
week_ago = datetime.utcnow() - timedelta(days=7)
268
week_ago_id = ObjectId.from_datetime(week_ago)
269
270
last_week_posts = await engine.find(
271
Post,
272
Post.id >= week_ago_id,
273
Post.id < yesterday_id
274
)
275
```
276
277
### Custom BSON Serialization
278
279
```python
280
from odmantic import WithBsonSerializer
281
from decimal import Decimal
282
import json
283
284
class CustomSerializable(WithBsonSerializer):
285
def __init__(self, data):
286
self.data = data
287
288
def __bson__(self):
289
# Custom serialization for BSON
290
return {
291
"type": "custom",
292
"data": json.dumps(self.data),
293
"version": 1
294
}
295
296
class DocumentWithCustom(Model):
297
name: str
298
custom_field: CustomSerializable
299
300
def custom_serialization_example():
301
# Create document with custom serialization
302
custom_data = CustomSerializable({"key": "value", "number": 42})
303
doc = DocumentWithCustom(name="test", custom_field=custom_data)
304
305
# When saved, custom_field will be serialized using __bson__ method
306
```
307
308
### Extended BSON Types Usage
309
310
```python
311
from odmantic.bson import Int64, Decimal128, Binary, Regex
312
from decimal import Decimal
313
314
class AdvancedDocument(Model):
315
name: str
316
large_number: Int64
317
precise_decimal: Decimal128
318
binary_data: Binary
319
pattern: Regex
320
321
def extended_types_example():
322
# Int64 for large integers
323
large_num = Int64(9223372036854775807)
324
325
# Decimal128 for high precision
326
precise = Decimal128(Decimal("99999.999999999999999999999999999"))
327
328
# Binary for byte data
329
binary = Binary(b"binary data here", subtype=0)
330
331
# Regex for pattern matching
332
regex = Regex(r"^[a-zA-Z]+$", flags=0)
333
334
doc = AdvancedDocument(
335
name="advanced",
336
large_number=large_num,
337
precise_decimal=precise,
338
binary_data=binary,
339
pattern=regex
340
)
341
```
342
343
### Working with Raw BSON
344
345
```python
346
import bson
347
from odmantic.bson import ObjectId
348
349
def raw_bson_examples():
350
# Converting to/from raw BSON
351
data = {
352
"_id": ObjectId(),
353
"name": "John",
354
"age": 30,
355
"balance": Decimal128("1234.56")
356
}
357
358
# Encode to BSON bytes
359
bson_bytes = bson.encode(data)
360
print(f"BSON size: {len(bson_bytes)} bytes")
361
362
# Decode from BSON bytes
363
decoded = bson.decode(bson_bytes)
364
print(f"Decoded: {decoded}")
365
366
# ODMantic handles this automatically, but you can work with raw BSON if needed
367
```
368
369
### Type Validation and Conversion
370
371
```python
372
from pydantic import field_validator
373
374
class ValidatedModel(Model):
375
id_field: ObjectId
376
int64_field: Int64
377
decimal_field: Decimal128
378
379
@field_validator('id_field', mode='before')
380
@classmethod
381
def validate_objectid(cls, v):
382
if isinstance(v, str):
383
if ObjectId.is_valid(v):
384
return ObjectId(v)
385
else:
386
raise ValueError('Invalid ObjectId format')
387
return v
388
389
@field_validator('decimal_field', mode='before')
390
@classmethod
391
def validate_decimal(cls, v):
392
if isinstance(v, (int, float, str)):
393
return Decimal128(str(v))
394
return v
395
396
def validation_example():
397
# Valid conversions
398
model = ValidatedModel(
399
id_field="507f1f77bcf86cd799439011", # String -> ObjectId
400
int64_field=123456789, # int -> Int64
401
decimal_field="123.456" # str -> Decimal128
402
)
403
404
print(f"ID: {model.id_field}")
405
print(f"Int64: {model.int64_field}")
406
print(f"Decimal: {model.decimal_field}")
407
```
408
409
### BSON Type Queries
410
411
```python
412
async def bson_type_queries(engine):
413
# Query by ObjectId
414
user_id = ObjectId("507f1f77bcf86cd799439011")
415
user = await engine.find_one(User, User.id == user_id)
416
417
# Query by ObjectId string (automatic conversion)
418
user = await engine.find_one(User, User.id == "507f1f77bcf86cd799439011")
419
420
# Query with Decimal128
421
products = await engine.find(Product, Product.price >= Decimal128("99.99"))
422
423
# Query with Int64
424
large_orders = await engine.find(Order, Order.total_cents >= Int64(10000))
425
426
# Regex queries
427
pattern_match = await engine.find(
428
User,
429
match(User.name, Regex(r"^John", flags=re.IGNORECASE))
430
)
431
```