(
rawdoc: Any, fields: Any, codec_options: CodecOptions[_DocumentType]
)
| 1179 | |
| 1180 | |
| 1181 | def _decode_selective( |
| 1182 | rawdoc: Any, fields: Any, codec_options: CodecOptions[_DocumentType] |
| 1183 | ) -> _DocumentType: |
| 1184 | if _raw_document_class(codec_options.document_class): |
| 1185 | # If document_class is RawBSONDocument, use vanilla dictionary for |
| 1186 | # decoding command response. |
| 1187 | doc: _DocumentType = {} # type:ignore[assignment] |
| 1188 | else: |
| 1189 | # Else, use the specified document_class. |
| 1190 | doc = codec_options.document_class() |
| 1191 | for key, value in rawdoc.items(): |
| 1192 | if key in fields: |
| 1193 | if fields[key] == 1: |
| 1194 | doc[key] = _bson_to_dict(rawdoc.raw, codec_options)[key] # type:ignore[index] |
| 1195 | else: |
| 1196 | doc[key] = _decode_selective( # type:ignore[index] |
| 1197 | value, fields[key], codec_options |
| 1198 | ) |
| 1199 | else: |
| 1200 | doc[key] = value # type:ignore[index] |
| 1201 | return doc |
| 1202 | |
| 1203 | |
| 1204 | def _array_of_documents_to_buffer(data: Union[memoryview, bytes]) -> bytes: |
no test coverage detected