Encode a document to BSON.
(
doc: Any, check_keys: bool, opts: CodecOptions[Any], top_level: bool = True
)
| 995 | |
| 996 | |
| 997 | def _dict_to_bson( |
| 998 | doc: Any, check_keys: bool, opts: CodecOptions[Any], top_level: bool = True |
| 999 | ) -> bytes: |
| 1000 | """Encode a document to BSON.""" |
| 1001 | if _raw_document_class(doc): |
| 1002 | return cast(bytes, doc.raw) |
| 1003 | try: |
| 1004 | elements = [] |
| 1005 | if top_level and "_id" in doc: |
| 1006 | elements.append(_name_value_to_bson(b"_id\x00", doc["_id"], check_keys, opts)) |
| 1007 | for key, value in doc.items(): |
| 1008 | if not top_level or key != "_id": |
| 1009 | try: |
| 1010 | elements.append(_element_to_bson(key, value, check_keys, opts)) |
| 1011 | except InvalidDocument as err: |
| 1012 | raise InvalidDocument(f"Invalid document: {err}", doc) from err |
| 1013 | except AttributeError: |
| 1014 | raise TypeError(f"encoder expected a mapping type but got: {doc!r}") from None |
| 1015 | |
| 1016 | encoded = b"".join(elements) |
| 1017 | return _PACK_INT(len(encoded) + 5) + encoded + b"\x00" |
| 1018 | |
| 1019 | |
| 1020 | if _USE_C: |