Official Python driver for MongoDB providing comprehensive tools for database operations, BSON handling, and GridFS file storage
—
BSON encoding/decoding functions and MongoDB-specific data types including ObjectId, Decimal128, Binary data, and timestamp handling.
Core functions for converting between Python documents and BSON (Binary JSON) format.
def encode(document, check_keys=False, codec_options=DEFAULT_CODEC_OPTIONS):
"""
Encode a Python document to BSON.
Parameters:
- document: Python dict or other mapping
- check_keys: if True, check for invalid key names
- codec_options: BSON codec options
Returns:
bytes: BSON-encoded document
"""
def decode(data, codec_options=DEFAULT_CODEC_OPTIONS):
"""
Decode BSON data to Python document.
Parameters:
- data: BSON bytes
- codec_options: BSON codec options
Returns:
dict: Decoded Python document
"""
def decode_all(data, codec_options=DEFAULT_CODEC_OPTIONS):
"""
Decode multiple BSON documents.
Parameters:
- data: BSON bytes containing multiple documents
- codec_options: BSON codec options
Returns:
list: List of decoded Python documents
"""
def decode_iter(data, codec_options=DEFAULT_CODEC_OPTIONS):
"""
Iterate over BSON documents in data.
Parameters:
- data: BSON bytes containing multiple documents
- codec_options: BSON codec options
Yields:
dict: Each decoded Python document
"""
def decode_file_iter(file_obj, codec_options=DEFAULT_CODEC_OPTIONS):
"""
Iterate over BSON documents from file.
Parameters:
- file_obj: file-like object containing BSON data
- codec_options: BSON codec options
Yields:
dict: Each decoded Python document
"""
def is_valid(bson):
"""
Check if data is valid BSON.
Parameters:
- bson: bytes to validate
Returns:
bool: True if valid BSON
"""
def has_c():
"""
Check if C extensions are available.
Returns:
bool: True if C extensions loaded
"""Container class for BSON data with encoding/decoding methods.
class BSON(bytes):
@classmethod
def encode(cls, document, check_keys=False, codec_options=DEFAULT_CODEC_OPTIONS):
"""
Encode document to BSON instance.
Parameters:
- document: Python dict or mapping
- check_keys: check for invalid key names
- codec_options: BSON codec options
Returns:
BSON: BSON instance containing encoded data
"""
def decode(self, codec_options=DEFAULT_CODEC_OPTIONS):
"""
Decode BSON data to Python document.
Parameters:
- codec_options: BSON codec options
Returns:
dict: Decoded Python document
"""MongoDB's unique identifier type with timestamp and machine information.
class ObjectId:
def __init__(self, oid=None):
"""
Create ObjectId from hex string, bytes, or generate new one.
Parameters:
- oid: 24-character hex string, 12 bytes, or None for new ObjectId
"""
@classmethod
def from_datetime(cls, generation_time):
"""
Create ObjectId from datetime.
Parameters:
- generation_time: datetime for ObjectId timestamp
Returns:
ObjectId: ObjectId with specified generation time
"""
@classmethod
def is_valid(cls, oid):
"""
Check if ObjectId is valid.
Parameters:
- oid: ObjectId candidate
Returns:
bool: True if valid ObjectId
"""
@property
def binary(self):
"""
ObjectId as 12 bytes.
Returns:
bytes: 12-byte ObjectId
"""
@property
def generation_time(self):
"""
Datetime when ObjectId was generated.
Returns:
datetime: Generation time (timezone-aware)
"""
def __str__(self):
"""
ObjectId as 24-character hex string.
Returns:
str: Hex representation
"""128-bit decimal number support for high-precision arithmetic.
class Decimal128:
def __init__(self, value):
"""
Create Decimal128 from string, int, float, or Decimal.
Parameters:
- value: numeric value to convert
"""
@classmethod
def from_bid(cls, value):
"""
Create from Binary Integer Decimal representation.
Parameters:
- value: 16-byte BID representation
Returns:
Decimal128: Decimal128 instance
"""
def to_decimal(self):
"""
Convert to Python Decimal.
Returns:
Decimal: Python decimal.Decimal instance
"""
@property
def bid(self):
"""
Binary Integer Decimal representation.
Returns:
bytes: 16-byte BID data
"""Support for binary data with different subtypes.
class Binary(bytes):
BINARY_SUBTYPE = 0
FUNCTION_SUBTYPE = 1
BINARY_SUBTYPE_OLD = 2
UUID_SUBTYPE = 3
UUID_SUBTYPE_OLD = 4
MD5_SUBTYPE = 5
ENCRYPTED_SUBTYPE = 6
COLUMN_SUBTYPE = 7
USER_DEFINED_SUBTYPE = 128
def __init__(self, data, subtype=BINARY_SUBTYPE):
"""
Create Binary data with subtype.
Parameters:
- data: binary data
- subtype: BSON binary subtype
"""
@property
def subtype(self):
"""
Binary subtype.
Returns:
int: BSON binary subtype
"""
class UUIDLegacy(Binary):
def __init__(self, uuid):
"""
Legacy UUID representation.
Parameters:
- uuid: UUID instance or bytes
"""
def as_uuid(self, uuid_representation=UuidRepresentation.UNSPECIFIED):
"""
Convert to UUID.
Parameters:
- uuid_representation: UUID representation format
Returns:
UUID: Python UUID instance
"""JavaScript code and regular expression types.
class Code(str):
def __init__(self, code, scope=None):
"""
JavaScript code with optional scope.
Parameters:
- code: JavaScript code string
- scope: optional scope dictionary
"""
@property
def scope(self):
"""
JavaScript scope variables.
Returns:
dict: Scope dictionary or None
"""
class Regex(str):
def __init__(self, pattern, flags=0):
"""
BSON regular expression.
Parameters:
- pattern: regex pattern string
- flags: regex flags
"""
@property
def pattern(self):
"""
Regular expression pattern.
Returns:
str: Regex pattern
"""
@property
def flags(self):
"""
Regular expression flags.
Returns:
int: Regex flags
"""
def try_compile(self):
"""
Compile to Python regex.
Returns:
Pattern: Compiled regex or None if invalid
"""Min/Max keys, timestamps, and database references.
class MinKey:
"""BSON MinKey - compares less than all other values."""
class MaxKey:
"""BSON MaxKey - compares greater than all other values."""
class Timestamp:
def __init__(self, time, inc):
"""
MongoDB timestamp.
Parameters:
- time: timestamp value (32-bit)
- inc: increment value (32-bit)
"""
@property
def time(self):
"""
Timestamp time component.
Returns:
int: Time value
"""
@property
def inc(self):
"""
Timestamp increment component.
Returns:
int: Increment value
"""
class DBRef:
def __init__(self, collection, id, database=None, **kwargs):
"""
Database reference.
Parameters:
- collection: collection name
- id: document identifier
- database: optional database name
"""
@property
def collection(self):
"""
Referenced collection name.
Returns:
str: Collection name
"""
@property
def id(self):
"""
Referenced document ID.
Returns:
Any: Document identifier
"""
@property
def database(self):
"""
Referenced database name.
Returns:
str: Database name or None
"""
class Int64(int):
"""64-bit integer type for platforms without native 64-bit support."""Configuration for BSON encoding/decoding behavior.
class CodecOptions:
def __init__(
self,
document_class=dict,
tz_aware=False,
uuid_representation=UuidRepresentation.UNSPECIFIED,
unicode_decode_error_handler='strict',
tzinfo=None,
type_registry=None,
datetime_conversion=DatetimeConversion.DATETIME
):
"""
BSON codec configuration.
Parameters:
- document_class: class for decoded documents
- tz_aware: timezone-aware datetime instances
- uuid_representation: UUID representation format
- unicode_decode_error_handler: Unicode error handling
- tzinfo: timezone for datetime instances
- type_registry: custom type registry
- datetime_conversion: datetime conversion mode
"""
DEFAULT_CODEC_OPTIONS: CodecOptionsimport bson
from bson import encode, decode, ObjectId
# Encode Python document to BSON
document = {"name": "Alice", "age": 30, "_id": ObjectId()}
bson_data = encode(document)
print(f"BSON size: {len(bson_data)} bytes")
# Decode BSON back to Python
decoded = decode(bson_data)
print(f"Decoded: {decoded}")
# Work with multiple documents
docs = [{"x": i} for i in range(3)]
bson_data = b''.join(encode(doc) for doc in docs)
# Decode all at once
all_docs = bson.decode_all(bson_data)
print(f"All docs: {all_docs}")
# Or iterate over them
for doc in bson.decode_iter(bson_data):
print(f"Doc: {doc}")from bson import ObjectId
from datetime import datetime
# Generate new ObjectId
oid = ObjectId()
print(f"ObjectId: {oid}")
print(f"Generated at: {oid.generation_time}")
# Create ObjectId from string
oid_str = str(oid)
oid2 = ObjectId(oid_str)
print(f"From string: {oid2}")
# Create ObjectId with specific timestamp
past_time = datetime(2023, 1, 1)
oid_past = ObjectId.from_datetime(past_time)
print(f"From datetime: {oid_past}")from bson import Decimal128
from decimal import Decimal
# Create from string for precision
dec128 = Decimal128("123.456789012345678901234567890")
print(f"Decimal128: {dec128}")
# Convert back to Python Decimal
py_decimal = dec128.to_decimal()
print(f"Python Decimal: {py_decimal}")
# Use in documents
document = {
"price": Decimal128("999.99"),
"tax_rate": Decimal128("0.0825")
}from bson import Binary
import uuid
# Store binary data
image_data = b'\x89PNG\r\n\x1a\n...' # PNG header
binary_field = Binary(image_data, Binary.BINARY_SUBTYPE)
# Store UUID
user_id = uuid.uuid4()
uuid_binary = Binary(user_id.bytes, Binary.UUID_SUBTYPE)
document = {
"image": binary_field,
"user_id": uuid_binary
}Install with Tessl CLI
npx tessl i tessl/pypi-pymongo