Decode a BSON array to python list.
(
data: Any, view: Any, position: int, obj_end: int, opts: CodecOptions[Any], element_name: str
)
| 332 | |
| 333 | |
| 334 | def _get_array( |
| 335 | data: Any, view: Any, position: int, obj_end: int, opts: CodecOptions[Any], element_name: str |
| 336 | ) -> Tuple[Any, int]: |
| 337 | """Decode a BSON array to python list.""" |
| 338 | size = _UNPACK_INT_FROM(data, position)[0] |
| 339 | end = position + size - 1 |
| 340 | if data[end] != 0: |
| 341 | raise InvalidBSON("bad eoo") |
| 342 | |
| 343 | position += 4 |
| 344 | end -= 1 |
| 345 | result: list[Any] = [] |
| 346 | |
| 347 | # Avoid doing global and attribute lookups in the loop. |
| 348 | append = result.append |
| 349 | index = data.index |
| 350 | getter = _ELEMENT_GETTER |
| 351 | decoder_map = opts.type_registry._decoder_map |
| 352 | |
| 353 | while position < end: |
| 354 | element_type = data[position] |
| 355 | # Just skip the keys. |
| 356 | position = index(b"\x00", position) + 1 |
| 357 | try: |
| 358 | value, position = getter[element_type]( |
| 359 | data, view, position, obj_end, opts, element_name |
| 360 | ) |
| 361 | except KeyError: |
| 362 | _raise_unknown_type(element_type, element_name) |
| 363 | |
| 364 | if decoder_map: |
| 365 | custom_decoder = decoder_map.get(type(value)) |
| 366 | if custom_decoder is not None: |
| 367 | value = custom_decoder(value) |
| 368 | |
| 369 | append(value) |
| 370 | |
| 371 | if position != end + 1: |
| 372 | raise InvalidBSON("bad array length") |
| 373 | return result, position + 1 |
| 374 | |
| 375 | |
| 376 | def _get_binary( |
nothing calls this directly
no test coverage detected