101
101
DEFAULT_CODEC_OPTIONS ,
102
102
CodecOptions ,
103
103
DatetimeConversion ,
104
- _DocumentType ,
105
104
_raw_document_class ,
106
105
)
107
106
from bson .datetime_ms import (
125
124
126
125
# Import some modules for type-checking only.
127
126
if TYPE_CHECKING :
128
- from array import array
129
- from mmap import mmap
127
+ from bson .typings import _DocumentIn , _DocumentType , _ReadableBuffer
130
128
131
129
try :
132
130
from bson import _cbson # type: ignore[attr-defined]
@@ -986,12 +984,8 @@ def _dict_to_bson(doc: Any, check_keys: bool, opts: CodecOptions, top_level: boo
986
984
_CODEC_OPTIONS_TYPE_ERROR = TypeError ("codec_options must be an instance of CodecOptions" )
987
985
988
986
989
- _DocumentIn = Mapping [str , Any ]
990
- _ReadableBuffer = Union [bytes , memoryview , "mmap" , "array" ]
991
-
992
-
993
987
def encode (
994
- document : _DocumentIn ,
988
+ document : " _DocumentIn" ,
995
989
check_keys : bool = False ,
996
990
codec_options : CodecOptions = DEFAULT_CODEC_OPTIONS ,
997
991
) -> bytes :
@@ -1022,8 +1016,8 @@ def encode(
1022
1016
1023
1017
1024
1018
def decode (
1025
- data : _ReadableBuffer , codec_options : "Optional[CodecOptions[_DocumentType]]" = None
1026
- ) -> _DocumentType :
1019
+ data : " _ReadableBuffer" , codec_options : "Optional[CodecOptions[_DocumentType]]" = None
1020
+ ) -> " _DocumentType" :
1027
1021
"""Decode BSON to a document.
1028
1022
1029
1023
By default, returns a BSON document represented as a Python
@@ -1056,11 +1050,13 @@ def decode(
1056
1050
return _bson_to_dict (data , opts )
1057
1051
1058
1052
1059
- def _decode_all (data : _ReadableBuffer , opts : "CodecOptions[_DocumentType]" ) -> List [_DocumentType ]:
1053
+ def _decode_all (
1054
+ data : "_ReadableBuffer" , opts : "CodecOptions[_DocumentType]"
1055
+ ) -> "List[_DocumentType]" :
1060
1056
"""Decode a BSON data to multiple documents."""
1061
1057
data , view = get_data_and_view (data )
1062
1058
data_len = len (data )
1063
- docs : List [_DocumentType ] = []
1059
+ docs : " List[_DocumentType]" = []
1064
1060
position = 0
1065
1061
end = data_len - 1
1066
1062
use_raw = _raw_document_class (opts .document_class )
@@ -1091,8 +1087,8 @@ def _decode_all(data: _ReadableBuffer, opts: "CodecOptions[_DocumentType]") -> L
1091
1087
1092
1088
1093
1089
def decode_all (
1094
- data : _ReadableBuffer , codec_options : "Optional[CodecOptions[_DocumentType]]" = None
1095
- ) -> List [_DocumentType ]:
1090
+ data : " _ReadableBuffer" , codec_options : "Optional[CodecOptions[_DocumentType]]" = None
1091
+ ) -> " List[_DocumentType]" :
1096
1092
"""Decode BSON data to multiple documents.
1097
1093
1098
1094
`data` must be a bytes-like object implementing the buffer protocol that
@@ -1213,7 +1209,7 @@ def _decode_all_selective(data: Any, codec_options: CodecOptions, fields: Any) -
1213
1209
# Decode documents for internal use.
1214
1210
from bson .raw_bson import RawBSONDocument
1215
1211
1216
- internal_codec_options = codec_options .with_options (
1212
+ internal_codec_options : CodecOptions [ RawBSONDocument ] = codec_options .with_options (
1217
1213
document_class = RawBSONDocument , type_registry = None
1218
1214
)
1219
1215
_doc = _bson_to_dict (data , internal_codec_options )
@@ -1228,7 +1224,7 @@ def _decode_all_selective(data: Any, codec_options: CodecOptions, fields: Any) -
1228
1224
1229
1225
def decode_iter (
1230
1226
data : bytes , codec_options : "Optional[CodecOptions[_DocumentType]]" = None
1231
- ) -> Iterator [_DocumentType ]:
1227
+ ) -> " Iterator[_DocumentType]" :
1232
1228
"""Decode BSON data to multiple documents as a generator.
1233
1229
1234
1230
Works similarly to the decode_all function, but yields one document at a
@@ -1264,7 +1260,7 @@ def decode_iter(
1264
1260
1265
1261
def decode_file_iter (
1266
1262
file_obj : Union [BinaryIO , IO ], codec_options : "Optional[CodecOptions[_DocumentType]]" = None
1267
- ) -> Iterator [_DocumentType ]:
1263
+ ) -> " Iterator[_DocumentType]" :
1268
1264
"""Decode bson data from a file to multiple documents as a generator.
1269
1265
1270
1266
Works similarly to the decode_all function, but reads from the file object
@@ -1325,7 +1321,7 @@ class BSON(bytes):
1325
1321
@classmethod
1326
1322
def encode (
1327
1323
cls : Type ["BSON" ],
1328
- document : _DocumentIn ,
1324
+ document : " _DocumentIn" ,
1329
1325
check_keys : bool = False ,
1330
1326
codec_options : CodecOptions = DEFAULT_CODEC_OPTIONS ,
1331
1327
) -> "BSON" :
@@ -1352,7 +1348,7 @@ def encode(
1352
1348
"""
1353
1349
return cls (encode (document , check_keys , codec_options ))
1354
1350
1355
- def decode (self , codec_options : "CodecOptions[_DocumentType]" = DEFAULT_CODEC_OPTIONS ) -> _DocumentType : # type: ignore[override,assignment]
1351
+ def decode (self , codec_options : "CodecOptions[_DocumentType]" = DEFAULT_CODEC_OPTIONS ) -> " _DocumentType" : # type: ignore[override,assignment]
1356
1352
"""Decode this BSON data.
1357
1353
1358
1354
By default, returns a BSON document represented as a Python
0 commit comments